Entity and methods

Project

Bases: Entity

A class representing a project.

Source code in digitalhub_core/entities/project/entity/_base.py

class Project(Entity):
    """
    A class representing a project.
    """

    ENTITY_TYPE = EntityTypes.PROJECT.value

    def __init__(
        self,
        name: str,
        kind: str,
        metadata: Metadata,
        spec: ProjectSpec,
        status: ProjectStatus,
        user: str | None = None,
        local: bool = False,
    ) -> None:
        super().__init__(kind, metadata, spec, status, user)
        self.id = name
        self.name = name
        self.key = f"store://{name}"
        self.spec: ProjectSpec
        self.status: ProjectStatus

        # Add attributes to be used in the to_dict method
        self._obj_attr.extend(["id", "name"])

        # Set client
        self._client = get_client(local)

        # Set context
        set_context(self)

    ##############################
    #  Save / Refresh / Export
    ##############################

    def save(self, update: bool = False) -> Project:
        """
        Save entity into backend.

        Parameters
        ----------
        update : bool
            If True, the object will be updated.

        Returns
        -------
        Project
            Entity saved.
        """
        obj = self._refresh_to_dict()

        if not update:
            new_obj = create_entity_api_base(self._client, self.ENTITY_TYPE, obj)
            new_obj["local"] = self._client.is_local()
            self._update_attributes(new_obj)
            return self

        self.metadata.updated = obj["metadata"]["updated"] = get_timestamp()
        new_obj = update_entity_api_base(self._client, self.ENTITY_TYPE, obj)
        new_obj["local"] = self._client.is_local()
        self._update_attributes(new_obj)
        return self

    def refresh(self) -> Project:
        """
        Refresh object from backend.

        Returns
        -------
        Project
            Project object.
        """
        new_obj = read_entity_api_base(self._client, self.ENTITY_TYPE, self.name)
        new_obj["local"] = self._client.is_local()
        self._update_attributes(new_obj)
        return self

    def export(self, filename: str | None = None) -> None:
        """
        Export object as a YAML file. If the objects are not embedded, the objects are
        exported as a YAML file.

        Parameters
        ----------
        filename : str
            Name of the export YAML file. If not specified, the default value is used.

        Returns
        -------
        None
        """
        obj = self._refresh_to_dict()

        if filename is None:
            filename = f"{self.kind}_{self.name}.yml"
        pth = Path(self.spec.context) / filename
        pth.parent.mkdir(parents=True, exist_ok=True)
        write_yaml(pth, obj)

        for entity_type in CTX_ENTITIES:
            entity_list = obj.get("spec", {}).get(entity_type, [])
            if not entity_list:
                continue
            self._export_not_embedded(entity_list, entity_type)

    def _refresh_to_dict(self) -> dict:
        """
        Try to refresh object to collect entities related to project.

        Returns
        -------
        dict
            Entity object in dictionary format.
        """
        try:
            return self.refresh().to_dict()
        except BackendError:
            return self.to_dict()

    def _export_not_embedded(self, entity_list: list, entity_type: str) -> None:
        """
        Export project objects if not embedded.

        Parameters
        ----------
        entity_list : list
            Entity list.

        Returns
        -------
        None
        """
        for entity in entity_list:
            if not entity["metadata"]["embedded"]:
                obj: dict = read_entity_api_ctx(entity["key"])
                ent = FUNC_MAP[entity_type](obj)
                ent.export()

    ##############################
    #  Static interface methods
    ##############################

    @staticmethod
    def _parse_dict(obj: dict, validate: bool = True) -> dict:
        """
        Get dictionary and parse it to a valid entity dictionary.

        Parameters
        ----------
        entity : str
            Entity type.
        obj : dict
            Dictionary to parse.

        Returns
        -------
        dict
            A dictionary containing the attributes of the entity instance.
        """
        name = build_name(obj.get("name"))
        kind = obj.get("kind")
        metadata = build_metadata(kind, **obj.get("metadata", {}))
        spec = build_spec(kind, validate=validate, **obj.get("spec", {}))
        status = build_status(kind, **obj.get("status", {}))
        user = obj.get("user")
        local = obj.get("local", False)
        return {
            "name": name,
            "kind": kind,
            "metadata": metadata,
            "spec": spec,
            "status": status,
            "user": user,
            "local": local,
        }

export(filename=None)

Export object as a YAML file. If the objects are not embedded, the objects are exported as a YAML file.

Parameters:

Name	Type	Description	Default
filename	str	Name of the export YAML file. If not specified, the default value is used.	None

Returns:

Type	Description
None

Source code in digitalhub_core/entities/project/entity/_base.py

def export(self, filename: str | None = None) -> None:
    """
    Export object as a YAML file. If the objects are not embedded, the objects are
    exported as a YAML file.

    Parameters
    ----------
    filename : str
        Name of the export YAML file. If not specified, the default value is used.

    Returns
    -------
    None
    """
    obj = self._refresh_to_dict()

    if filename is None:
        filename = f"{self.kind}_{self.name}.yml"
    pth = Path(self.spec.context) / filename
    pth.parent.mkdir(parents=True, exist_ok=True)
    write_yaml(pth, obj)

    for entity_type in CTX_ENTITIES:
        entity_list = obj.get("spec", {}).get(entity_type, [])
        if not entity_list:
            continue
        self._export_not_embedded(entity_list, entity_type)

refresh()

Refresh object from backend.

Returns:

Type	Description
Project	Project object.

Source code in digitalhub_core/entities/project/entity/_base.py

def refresh(self) -> Project:
    """
    Refresh object from backend.

    Returns
    -------
    Project
        Project object.
    """
    new_obj = read_entity_api_base(self._client, self.ENTITY_TYPE, self.name)
    new_obj["local"] = self._client.is_local()
    self._update_attributes(new_obj)
    return self

save(update=False)

Save entity into backend.

Parameters:

Name	Type	Description	Default
update	bool	If True, the object will be updated.	False

Returns:

Type	Description
Project	Entity saved.

Source code in digitalhub_core/entities/project/entity/_base.py

def save(self, update: bool = False) -> Project:
    """
    Save entity into backend.

    Parameters
    ----------
    update : bool
        If True, the object will be updated.

    Returns
    -------
    Project
        Entity saved.
    """
    obj = self._refresh_to_dict()

    if not update:
        new_obj = create_entity_api_base(self._client, self.ENTITY_TYPE, obj)
        new_obj["local"] = self._client.is_local()
        self._update_attributes(new_obj)
        return self

    self.metadata.updated = obj["metadata"]["updated"] = get_timestamp()
    new_obj = update_entity_api_base(self._client, self.ENTITY_TYPE, obj)
    new_obj["local"] = self._client.is_local()
    self._update_attributes(new_obj)
    return self

ProjectCore

Bases: Project

Source code in digitalhub_core/entities/project/entity/core.py

class ProjectCore(Project):
    ##############################
    #  Artifacts
    ##############################

    def new_artifact(
        self,
        name: str,
        kind: str,
        uuid: str | None = None,
        description: str | None = None,
        labels: list[str] | None = None,
        embedded: bool = True,
        path: str | None = None,
        **kwargs,
    ) -> Artifact:
        """
        Create a new object.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        uuid : str
            ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
        description : str
            Description of the object (human readable).
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object spec must be embedded in project spec.
        path : str
            Object path on local file system or remote storage. It is also the destination path of upload() method.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Artifact
            Object instance.

        Examples
        --------
        >>> obj = project.new_artifact(name="my-artifact",
        >>>                            kind="artifact",
        >>>                            path="s3://my-bucket/my-key")
        """
        obj = new_artifact(
            project=self.name,
            name=name,
            kind=kind,
            uuid=uuid,
            description=description,
            labels=labels,
            embedded=embedded,
            path=path,
            **kwargs,
        )
        self.refresh()
        return obj

    def log_artifact(
        self,
        name: str,
        kind: str,
        source: str,
        path: str | None = None,
        **kwargs,
    ) -> Artifact:
        """
        Create and upload an object.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        source : str
            Artifact location on local path.
        path : str
            Destination path of the artifact. If not provided, it's generated.
        **kwargs : dict
            New artifact spec parameters.

        Returns
        -------
        Artifact
            Object instance.

        Examples
        --------
        >>> obj = project.log_artifact(name="my-artifact",
        >>>                            kind="artifact",
        >>>                            source="./local-path")
        """
        obj = log_artifact(
            project=self.name,
            name=name,
            kind=kind,
            source=source,
            path=path,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_artifact(
        self,
        identifier: str,
        entity_id: str | None = None,
        **kwargs,
    ) -> Artifact:
        """
        Get object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Artifact
            Object instance.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_artifact("store://my-artifact-key")

        Using entity name:
        >>> obj = project.get_artifact("my-artifact-name"
        >>>                            entity_id="my-artifact-id")
        """
        obj = get_artifact(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_artifact_versions(
        self,
        identifier: str,
        **kwargs,
    ) -> list[Artifact]:
        """
        Get object versions from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Artifact]
            List of object instances.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_artifact_versions("store://my-artifact-key")

        Using entity name:
        >>> obj = project.get_artifact_versions("my-artifact-name")
        """
        return get_artifact_versions(identifier, project=self.name, **kwargs)

    def list_artifacts(self, **kwargs) -> list[Artifact]:
        """
        List all latest version objects from backend.

        Parameters
        ----------
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Artifact]
            List of object instances.

        Examples
        --------
        >>> objs = project.list_artifacts()
        """
        return list_artifacts(self.name, **kwargs)

    def import_artifact(
        self,
        file: str,
        **kwargs,
    ) -> Artifact:
        """
        Import object from a YAML file.

        Parameters
        ----------
        file : str
            Path to YAML file.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Artifact
            Object instance.

        Examples
        --------
        >>> obj = project.import_artifact("my-artifact.yaml")
        """
        return import_artifact(file, **kwargs)

    def update_artifact(self, entity: Artifact) -> Artifact:
        """
        Update object. Note that object spec are immutable.

        Parameters
        ----------
        entity : Artifact
            Object to update.

        Returns
        -------
        Artifact
            Entity updated.

        Examples
        --------
        >>> obj = project.update_artifact(obj)
        """
        if entity.project != self.name:
            raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
        return update_artifact(entity)

    def delete_artifact(
        self,
        identifier: str,
        entity_id: str | None = None,
        delete_all_versions: bool = False,
        **kwargs,
    ) -> None:
        """
        Delete object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        dict
            Response from backend.

        Examples
        --------
        If delete_all_versions is False:
        >>> project.delete_artifact("store://my-artifact-key")

        Otherwise:
        >>> project.delete_artifact("my-artifact-name",
        >>>                         delete_all_versions=True)
        """
        delete_artifact(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            **kwargs,
        )
        self.refresh()

    ##############################
    #  Functions
    ##############################

    def new_function(
        self,
        name: str,
        kind: str,
        uuid: str | None = None,
        description: str | None = None,
        labels: list[str] | None = None,
        embedded: bool = True,
        **kwargs,
    ) -> Function:
        """
        Create a new object.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        uuid : str
            ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
        description : str
            Description of the object (human readable).
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object spec must be embedded in project spec.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Function
            Object instance.

        Examples
        --------
        >>> obj = project.new_function(name="my-function",
        >>>                            kind="python",
        >>>                            code_src="function.py",
        >>>                            handler="function-handler")
        """
        obj = new_function(
            project=self.name,
            name=name,
            kind=kind,
            uuid=uuid,
            description=description,
            labels=labels,
            embedded=embedded,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_function(
        self,
        identifier: str,
        entity_id: str | None = None,
        **kwargs,
    ) -> Function:
        """
        Get object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Function
            Object instance.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_function("store://my-function-key")

        Using entity name:
        >>> obj = project.get_function("my-function-name"
        >>>                            entity_id="my-function-id")
        """
        obj = get_function(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_function_versions(
        self,
        identifier: str,
        **kwargs,
    ) -> list[Function]:
        """
        Get object versions from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Function]
            List of object instances.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_function_versions("store://my-function-key")

        Using entity name:
        >>> obj = project.get_function_versions("my-function-name")
        """
        return get_function_versions(identifier, project=self.name, **kwargs)

    def list_functions(self, **kwargs) -> list[Function]:
        """
        List all latest version objects from backend.

        Parameters
        ----------
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Function]
            List of object instances.

        Examples
        --------
        >>> objs = project.list_functions()
        """
        return list_functions(self.name, **kwargs)

    def import_function(
        self,
        file: str,
        **kwargs,
    ) -> Function:
        """
        Import object from a YAML file.

        Parameters
        ----------
        file : str
            Path to YAML file.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Function
            Object instance.

        Examples
        --------
        >>> obj = project.import_function("my-function.yaml")
        """
        return import_function(file, **kwargs)

    def update_function(self, entity: Function) -> Function:
        """
        Update object. Note that object spec are immutable.

        Parameters
        ----------
        entity : Function
            Object to update.

        Returns
        -------
        Function
            Entity updated.

        Examples
        --------
        >>> obj = project.update_function(obj)
        """
        if entity.project != self.name:
            raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
        return update_function(entity)

    def delete_function(
        self,
        identifier: str,
        entity_id: str | None = None,
        delete_all_versions: bool = False,
        cascade: bool = True,
        **kwargs,
    ) -> None:
        """
        Delete object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
        cascade : bool
            Cascade delete.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        dict
            Response from backend.

        Examples
        --------
        If delete_all_versions is False:
        >>> project.delete_function("store://my-function-key")

        Otherwise:
        >>> project.delete_function("my-function-name",
        >>>                         delete_all_versions=True)
        """
        delete_function(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            cascade=cascade,
            **kwargs,
        )
        self.refresh()

    ##############################
    #  Workflows
    ##############################

    def new_workflow(
        self,
        name: str,
        kind: str,
        uuid: str | None = None,
        description: str | None = None,
        labels: list[str] | None = None,
        embedded: bool = True,
        **kwargs,
    ) -> Workflow:
        """
        Create a new object.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        uuid : str
            ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
        description : str
            Description of the object (human readable).
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object spec must be embedded in project spec.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Workflow
            Object instance.

        Examples
        --------
        >>> obj = project.new_workflow(name="my-workflow",
        >>>                            kind="kfp",
        >>>                            code_src="pipeline.py",
        >>>                            handler="pipeline-handler")
        """
        obj = new_workflow(
            project=self.name,
            name=name,
            kind=kind,
            uuid=uuid,
            description=description,
            labels=labels,
            embedded=embedded,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_workflow(
        self,
        identifier: str,
        entity_id: str | None = None,
        **kwargs,
    ) -> Workflow:
        """
        Get object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Workflow
            Object instance.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_workflow("store://my-workflow-key")

        Using entity name:
        >>> obj = project.get_workflow("my-workflow-name"
        >>>                            entity_id="my-workflow-id")
        """
        obj = get_workflow(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_workflow_versions(
        self,
        identifier: str,
        **kwargs,
    ) -> list[Workflow]:
        """
        Get object versions from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Workflow]
            List of object instances.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_workflow_versions("store://my-workflow-key")

        Using entity name:
        >>> obj = project.get_workflow_versions("my-workflow-name")
        """
        return get_workflow_versions(identifier, project=self.name, **kwargs)

    def list_workflows(self, **kwargs) -> list[Workflow]:
        """
        List all latest version objects from backend.

        Parameters
        ----------
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Workflow]
            List of object instances.

        Examples
        --------
        >>> objs = project.list_workflows()
        """
        return list_workflows(self.name, **kwargs)

    def import_workflow(
        self,
        file: str,
        **kwargs,
    ) -> Workflow:
        """
        Import object from a YAML file.

        Parameters
        ----------
        file : str
            Path to YAML file.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Workflow
            Object instance.

        Examples
        --------
        >>> obj = project.import_workflow("my-workflow.yaml")
        """
        return import_workflow(file, **kwargs)

    def update_workflow(self, entity: Workflow) -> Workflow:
        """
        Update object. Note that object spec are immutable.

        Parameters
        ----------
        entity : Workflow
            Object to update.

        Returns
        -------
        Workflow
            Entity updated.

        Examples
        --------
        >>> obj = project.update_workflow(obj)
        """
        if entity.project != self.name:
            raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
        return update_workflow(entity)

    def delete_workflow(
        self,
        identifier: str,
        entity_id: str | None = None,
        delete_all_versions: bool = False,
        cascade: bool = True,
        **kwargs,
    ) -> None:
        """
        Delete object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
        cascade : bool
            Cascade delete.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        dict
            Response from backend.

        Examples
        --------
        If delete_all_versions is False:
        >>> project.delete_workflow("store://my-workflow-key")

        Otherwise:
        >>> project.delete_workflow("my-workflow-name",
        >>>                         delete_all_versions=True)
        """
        delete_workflow(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            cascade=cascade,
            **kwargs,
        )
        self.refresh()

    ##############################
    #  Secrets
    ##############################

    def new_secret(
        self,
        name: str,
        uuid: str | None = None,
        description: str | None = None,
        labels: list[str] | None = None,
        embedded: bool = True,
        secret_value: str | None = None,
        **kwargs,
    ) -> Secret:
        """
        Create a new object.

        Parameters
        ----------
        name : str
            Object name.
        uuid : str
            ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
        description : str
            Description of the object (human readable).
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object spec must be embedded in project spec.
        secret_value : str
            Value of the secret.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Secret
            Object instance.

        Examples
        --------
        >>> obj = project.new_secret(name="my-secret",
        >>>                          secret_value="my-secret-value")
        """
        obj = new_secret(
            project=self.name,
            name=name,
            uuid=uuid,
            description=description,
            labels=labels,
            embedded=embedded,
            secret_value=secret_value,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_secret(
        self,
        identifier: str,
        entity_id: str | None = None,
        **kwargs,
    ) -> Secret:
        """
        Get object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Secret
            Object instance.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_secret("store://my-secret-key")

        Using entity name:
        >>> obj = project.get_secret("my-secret-name"
        >>>                          entity_id="my-secret-id")
        """
        obj = get_secret(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_secret_versions(
        self,
        identifier: str,
        **kwargs,
    ) -> list[Secret]:
        """
        Get object versions from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Secret]
            List of object instances.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_secret_versions("store://my-secret-key")

        Using entity name:
        >>> obj = project.get_secret_versions("my-secret-name")
        """
        return get_secret_versions(identifier, project=self.name, **kwargs)

    def list_secrets(self, **kwargs) -> list[Secret]:
        """
        List all latest version objects from backend.

        Parameters
        ----------
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Secret]
            List of object instances.

        Examples
        --------
        >>> objs = project.list_secrets()
        """
        return list_secrets(self.name, **kwargs)

    def import_secret(
        self,
        file: str,
        **kwargs,
    ) -> Secret:
        """
        Import object from a YAML file.

        Parameters
        ----------
        file : str
            Path to YAML file.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Secret
            Object instance.

        Examples
        --------
        >>> obj = project.import_secret("my-secret.yaml")
        """
        return import_secret(file, **kwargs)

    def update_secret(self, entity: Secret) -> Secret:
        """
        Update object. Note that object spec are immutable.

        Parameters
        ----------
        entity : Secret
            Object to update.

        Returns
        -------
        Secret
            Entity updated.

        Examples
        --------
        >>> obj = project.update_secret(obj)
        """
        if entity.project != self.name:
            raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
        return update_secret(entity)

    def delete_secret(
        self,
        identifier: str,
        entity_id: str | None = None,
        delete_all_versions: bool = False,
        **kwargs,
    ) -> None:
        """
        Delete object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        dict
            Response from backend.

        Examples
        --------
        If delete_all_versions is False:
        >>> project.delete_secret("store://my-secret-key")

        Otherwise:
        >>> project.delete_secret("my-secret-name",
        >>>                       delete_all_versions=True)
        """
        delete_secret(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            **kwargs,
        )
        self.refresh()

delete_artifact(identifier, entity_id=None, delete_all_versions=False, **kwargs)

Delete object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.	False
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
dict	Response from backend.

Examples:

If delete_all_versions is False:

>>> project.delete_artifact("store://my-artifact-key")

Otherwise:

>>> project.delete_artifact("my-artifact-name",
>>>                         delete_all_versions=True)

Source code in digitalhub_core/entities/project/entity/core.py

def delete_artifact(
    self,
    identifier: str,
    entity_id: str | None = None,
    delete_all_versions: bool = False,
    **kwargs,
) -> None:
    """
    Delete object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    dict
        Response from backend.

    Examples
    --------
    If delete_all_versions is False:
    >>> project.delete_artifact("store://my-artifact-key")

    Otherwise:
    >>> project.delete_artifact("my-artifact-name",
    >>>                         delete_all_versions=True)
    """
    delete_artifact(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        delete_all_versions=delete_all_versions,
        **kwargs,
    )
    self.refresh()

delete_function(identifier, entity_id=None, delete_all_versions=False, cascade=True, **kwargs)

Delete object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.	False
cascade	bool	Cascade delete.	True
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
dict	Response from backend.

Examples:

If delete_all_versions is False:

>>> project.delete_function("store://my-function-key")

Otherwise:

>>> project.delete_function("my-function-name",
>>>                         delete_all_versions=True)

Source code in digitalhub_core/entities/project/entity/core.py

def delete_function(
    self,
    identifier: str,
    entity_id: str | None = None,
    delete_all_versions: bool = False,
    cascade: bool = True,
    **kwargs,
) -> None:
    """
    Delete object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
    cascade : bool
        Cascade delete.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    dict
        Response from backend.

    Examples
    --------
    If delete_all_versions is False:
    >>> project.delete_function("store://my-function-key")

    Otherwise:
    >>> project.delete_function("my-function-name",
    >>>                         delete_all_versions=True)
    """
    delete_function(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        delete_all_versions=delete_all_versions,
        cascade=cascade,
        **kwargs,
    )
    self.refresh()

delete_secret(identifier, entity_id=None, delete_all_versions=False, **kwargs)

Delete object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.	False
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
dict	Response from backend.

Examples:

If delete_all_versions is False:

>>> project.delete_secret("store://my-secret-key")

Otherwise:

>>> project.delete_secret("my-secret-name",
>>>                       delete_all_versions=True)

Source code in digitalhub_core/entities/project/entity/core.py

def delete_secret(
    self,
    identifier: str,
    entity_id: str | None = None,
    delete_all_versions: bool = False,
    **kwargs,
) -> None:
    """
    Delete object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    dict
        Response from backend.

    Examples
    --------
    If delete_all_versions is False:
    >>> project.delete_secret("store://my-secret-key")

    Otherwise:
    >>> project.delete_secret("my-secret-name",
    >>>                       delete_all_versions=True)
    """
    delete_secret(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        delete_all_versions=delete_all_versions,
        **kwargs,
    )
    self.refresh()

delete_workflow(identifier, entity_id=None, delete_all_versions=False, cascade=True, **kwargs)

Delete object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.	False
cascade	bool	Cascade delete.	True
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
dict	Response from backend.

Examples:

If delete_all_versions is False:

>>> project.delete_workflow("store://my-workflow-key")

Otherwise:

>>> project.delete_workflow("my-workflow-name",
>>>                         delete_all_versions=True)

Source code in digitalhub_core/entities/project/entity/core.py

def delete_workflow(
    self,
    identifier: str,
    entity_id: str | None = None,
    delete_all_versions: bool = False,
    cascade: bool = True,
    **kwargs,
) -> None:
    """
    Delete object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
    cascade : bool
        Cascade delete.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    dict
        Response from backend.

    Examples
    --------
    If delete_all_versions is False:
    >>> project.delete_workflow("store://my-workflow-key")

    Otherwise:
    >>> project.delete_workflow("my-workflow-name",
    >>>                         delete_all_versions=True)
    """
    delete_workflow(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        delete_all_versions=delete_all_versions,
        cascade=cascade,
        **kwargs,
    )
    self.refresh()

get_artifact(identifier, entity_id=None, **kwargs)

Get object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Artifact	Object instance.

Examples:

Using entity key:

>>> obj = project.get_artifact("store://my-artifact-key")

Using entity name:

>>> obj = project.get_artifact("my-artifact-name"
>>>                            entity_id="my-artifact-id")

Source code in digitalhub_core/entities/project/entity/core.py

def get_artifact(
    self,
    identifier: str,
    entity_id: str | None = None,
    **kwargs,
) -> Artifact:
    """
    Get object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Artifact
        Object instance.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_artifact("store://my-artifact-key")

    Using entity name:
    >>> obj = project.get_artifact("my-artifact-name"
    >>>                            entity_id="my-artifact-id")
    """
    obj = get_artifact(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

get_artifact_versions(identifier, **kwargs)

Get object versions from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Artifact]	List of object instances.

Examples:

Using entity key:

>>> obj = project.get_artifact_versions("store://my-artifact-key")

Using entity name:

>>> obj = project.get_artifact_versions("my-artifact-name")

Source code in digitalhub_core/entities/project/entity/core.py

def get_artifact_versions(
    self,
    identifier: str,
    **kwargs,
) -> list[Artifact]:
    """
    Get object versions from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Artifact]
        List of object instances.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_artifact_versions("store://my-artifact-key")

    Using entity name:
    >>> obj = project.get_artifact_versions("my-artifact-name")
    """
    return get_artifact_versions(identifier, project=self.name, **kwargs)

get_function(identifier, entity_id=None, **kwargs)

Get object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Function	Object instance.

Examples:

Using entity key:

>>> obj = project.get_function("store://my-function-key")

Using entity name:

>>> obj = project.get_function("my-function-name"
>>>                            entity_id="my-function-id")

Source code in digitalhub_core/entities/project/entity/core.py

def get_function(
    self,
    identifier: str,
    entity_id: str | None = None,
    **kwargs,
) -> Function:
    """
    Get object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Function
        Object instance.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_function("store://my-function-key")

    Using entity name:
    >>> obj = project.get_function("my-function-name"
    >>>                            entity_id="my-function-id")
    """
    obj = get_function(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

get_function_versions(identifier, **kwargs)

Get object versions from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Function]	List of object instances.

Examples:

Using entity key:

>>> obj = project.get_function_versions("store://my-function-key")

Using entity name:

>>> obj = project.get_function_versions("my-function-name")

Source code in digitalhub_core/entities/project/entity/core.py

def get_function_versions(
    self,
    identifier: str,
    **kwargs,
) -> list[Function]:
    """
    Get object versions from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Function]
        List of object instances.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_function_versions("store://my-function-key")

    Using entity name:
    >>> obj = project.get_function_versions("my-function-name")
    """
    return get_function_versions(identifier, project=self.name, **kwargs)

get_secret(identifier, entity_id=None, **kwargs)

Get object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Secret	Object instance.

Examples:

Using entity key:

>>> obj = project.get_secret("store://my-secret-key")

Using entity name:

>>> obj = project.get_secret("my-secret-name"
>>>                          entity_id="my-secret-id")

Source code in digitalhub_core/entities/project/entity/core.py

def get_secret(
    self,
    identifier: str,
    entity_id: str | None = None,
    **kwargs,
) -> Secret:
    """
    Get object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Secret
        Object instance.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_secret("store://my-secret-key")

    Using entity name:
    >>> obj = project.get_secret("my-secret-name"
    >>>                          entity_id="my-secret-id")
    """
    obj = get_secret(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

get_secret_versions(identifier, **kwargs)

Get object versions from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Secret]	List of object instances.

Examples:

Using entity key:

>>> obj = project.get_secret_versions("store://my-secret-key")

Using entity name:

>>> obj = project.get_secret_versions("my-secret-name")

Source code in digitalhub_core/entities/project/entity/core.py

def get_secret_versions(
    self,
    identifier: str,
    **kwargs,
) -> list[Secret]:
    """
    Get object versions from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Secret]
        List of object instances.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_secret_versions("store://my-secret-key")

    Using entity name:
    >>> obj = project.get_secret_versions("my-secret-name")
    """
    return get_secret_versions(identifier, project=self.name, **kwargs)

get_workflow(identifier, entity_id=None, **kwargs)

Get object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Workflow	Object instance.

Examples:

Using entity key:

>>> obj = project.get_workflow("store://my-workflow-key")

Using entity name:

>>> obj = project.get_workflow("my-workflow-name"
>>>                            entity_id="my-workflow-id")

Source code in digitalhub_core/entities/project/entity/core.py

def get_workflow(
    self,
    identifier: str,
    entity_id: str | None = None,
    **kwargs,
) -> Workflow:
    """
    Get object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Workflow
        Object instance.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_workflow("store://my-workflow-key")

    Using entity name:
    >>> obj = project.get_workflow("my-workflow-name"
    >>>                            entity_id="my-workflow-id")
    """
    obj = get_workflow(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

get_workflow_versions(identifier, **kwargs)

Get object versions from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Workflow]	List of object instances.

Examples:

Using entity key:

>>> obj = project.get_workflow_versions("store://my-workflow-key")

Using entity name:

>>> obj = project.get_workflow_versions("my-workflow-name")

Source code in digitalhub_core/entities/project/entity/core.py

def get_workflow_versions(
    self,
    identifier: str,
    **kwargs,
) -> list[Workflow]:
    """
    Get object versions from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Workflow]
        List of object instances.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_workflow_versions("store://my-workflow-key")

    Using entity name:
    >>> obj = project.get_workflow_versions("my-workflow-name")
    """
    return get_workflow_versions(identifier, project=self.name, **kwargs)

import_artifact(file, **kwargs)

Import object from a YAML file.

Parameters:

Name	Type	Description	Default
file	str	Path to YAML file.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Artifact	Object instance.

Examples:

>>> obj = project.import_artifact("my-artifact.yaml")

Source code in digitalhub_core/entities/project/entity/core.py

def import_artifact(
    self,
    file: str,
    **kwargs,
) -> Artifact:
    """
    Import object from a YAML file.

    Parameters
    ----------
    file : str
        Path to YAML file.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Artifact
        Object instance.

    Examples
    --------
    >>> obj = project.import_artifact("my-artifact.yaml")
    """
    return import_artifact(file, **kwargs)

import_function(file, **kwargs)

Import object from a YAML file.

Parameters:

Name	Type	Description	Default
file	str	Path to YAML file.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Function	Object instance.

Examples:

>>> obj = project.import_function("my-function.yaml")

Source code in digitalhub_core/entities/project/entity/core.py

def import_function(
    self,
    file: str,
    **kwargs,
) -> Function:
    """
    Import object from a YAML file.

    Parameters
    ----------
    file : str
        Path to YAML file.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Function
        Object instance.

    Examples
    --------
    >>> obj = project.import_function("my-function.yaml")
    """
    return import_function(file, **kwargs)

import_secret(file, **kwargs)

Import object from a YAML file.

Parameters:

Name	Type	Description	Default
file	str	Path to YAML file.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Secret	Object instance.

Examples:

>>> obj = project.import_secret("my-secret.yaml")

Source code in digitalhub_core/entities/project/entity/core.py

def import_secret(
    self,
    file: str,
    **kwargs,
) -> Secret:
    """
    Import object from a YAML file.

    Parameters
    ----------
    file : str
        Path to YAML file.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Secret
        Object instance.

    Examples
    --------
    >>> obj = project.import_secret("my-secret.yaml")
    """
    return import_secret(file, **kwargs)

import_workflow(file, **kwargs)

Import object from a YAML file.

Parameters:

Name	Type	Description	Default
file	str	Path to YAML file.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Workflow	Object instance.

Examples:

>>> obj = project.import_workflow("my-workflow.yaml")

Source code in digitalhub_core/entities/project/entity/core.py

def import_workflow(
    self,
    file: str,
    **kwargs,
) -> Workflow:
    """
    Import object from a YAML file.

    Parameters
    ----------
    file : str
        Path to YAML file.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Workflow
        Object instance.

    Examples
    --------
    >>> obj = project.import_workflow("my-workflow.yaml")
    """
    return import_workflow(file, **kwargs)

list_artifacts(**kwargs)

List all latest version objects from backend.

Parameters:

Name	Type	Description	Default
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Artifact]	List of object instances.

Examples:

>>> objs = project.list_artifacts()

Source code in digitalhub_core/entities/project/entity/core.py

def list_artifacts(self, **kwargs) -> list[Artifact]:
    """
    List all latest version objects from backend.

    Parameters
    ----------
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Artifact]
        List of object instances.

    Examples
    --------
    >>> objs = project.list_artifacts()
    """
    return list_artifacts(self.name, **kwargs)

list_functions(**kwargs)

List all latest version objects from backend.

Parameters:

Name	Type	Description	Default
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Function]	List of object instances.

Examples:

>>> objs = project.list_functions()

Source code in digitalhub_core/entities/project/entity/core.py

def list_functions(self, **kwargs) -> list[Function]:
    """
    List all latest version objects from backend.

    Parameters
    ----------
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Function]
        List of object instances.

    Examples
    --------
    >>> objs = project.list_functions()
    """
    return list_functions(self.name, **kwargs)

list_secrets(**kwargs)

List all latest version objects from backend.

Parameters:

Name	Type	Description	Default
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Secret]	List of object instances.

Examples:

>>> objs = project.list_secrets()

Source code in digitalhub_core/entities/project/entity/core.py

def list_secrets(self, **kwargs) -> list[Secret]:
    """
    List all latest version objects from backend.

    Parameters
    ----------
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Secret]
        List of object instances.

    Examples
    --------
    >>> objs = project.list_secrets()
    """
    return list_secrets(self.name, **kwargs)

list_workflows(**kwargs)

List all latest version objects from backend.

Parameters:

Name	Type	Description	Default
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Workflow]	List of object instances.

Examples:

>>> objs = project.list_workflows()

Source code in digitalhub_core/entities/project/entity/core.py

def list_workflows(self, **kwargs) -> list[Workflow]:
    """
    List all latest version objects from backend.

    Parameters
    ----------
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Workflow]
        List of object instances.

    Examples
    --------
    >>> objs = project.list_workflows()
    """
    return list_workflows(self.name, **kwargs)

log_artifact(name, kind, source, path=None, **kwargs)

Create and upload an object.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
source	str	Artifact location on local path.	required
path	str	Destination path of the artifact. If not provided, it's generated.	None
**kwargs	dict	New artifact spec parameters.	{}

Returns:

Type	Description
Artifact	Object instance.

Examples:

>>> obj = project.log_artifact(name="my-artifact",
>>>                            kind="artifact",
>>>                            source="./local-path")

Source code in digitalhub_core/entities/project/entity/core.py

def log_artifact(
    self,
    name: str,
    kind: str,
    source: str,
    path: str | None = None,
    **kwargs,
) -> Artifact:
    """
    Create and upload an object.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    source : str
        Artifact location on local path.
    path : str
        Destination path of the artifact. If not provided, it's generated.
    **kwargs : dict
        New artifact spec parameters.

    Returns
    -------
    Artifact
        Object instance.

    Examples
    --------
    >>> obj = project.log_artifact(name="my-artifact",
    >>>                            kind="artifact",
    >>>                            source="./local-path")
    """
    obj = log_artifact(
        project=self.name,
        name=name,
        kind=kind,
        source=source,
        path=path,
        **kwargs,
    )
    self.refresh()
    return obj

new_artifact(name, kind, uuid=None, description=None, labels=None, embedded=True, path=None, **kwargs)

Create a new object.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
uuid	str	ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).	None
description	str	Description of the object (human readable).	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object spec must be embedded in project spec.	True
path	str	Object path on local file system or remote storage. It is also the destination path of upload() method.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Artifact	Object instance.

Examples:

>>> obj = project.new_artifact(name="my-artifact",
>>>                            kind="artifact",
>>>                            path="s3://my-bucket/my-key")

Source code in digitalhub_core/entities/project/entity/core.py

def new_artifact(
    self,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    path: str | None = None,
    **kwargs,
) -> Artifact:
    """
    Create a new object.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
    description : str
        Description of the object (human readable).
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object spec must be embedded in project spec.
    path : str
        Object path on local file system or remote storage. It is also the destination path of upload() method.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Artifact
        Object instance.

    Examples
    --------
    >>> obj = project.new_artifact(name="my-artifact",
    >>>                            kind="artifact",
    >>>                            path="s3://my-bucket/my-key")
    """
    obj = new_artifact(
        project=self.name,
        name=name,
        kind=kind,
        uuid=uuid,
        description=description,
        labels=labels,
        embedded=embedded,
        path=path,
        **kwargs,
    )
    self.refresh()
    return obj

new_function(name, kind, uuid=None, description=None, labels=None, embedded=True, **kwargs)

Create a new object.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
uuid	str	ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).	None
description	str	Description of the object (human readable).	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object spec must be embedded in project spec.	True
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Function	Object instance.

Examples:

>>> obj = project.new_function(name="my-function",
>>>                            kind="python",
>>>                            code_src="function.py",
>>>                            handler="function-handler")

Source code in digitalhub_core/entities/project/entity/core.py

def new_function(
    self,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    **kwargs,
) -> Function:
    """
    Create a new object.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
    description : str
        Description of the object (human readable).
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object spec must be embedded in project spec.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Function
        Object instance.

    Examples
    --------
    >>> obj = project.new_function(name="my-function",
    >>>                            kind="python",
    >>>                            code_src="function.py",
    >>>                            handler="function-handler")
    """
    obj = new_function(
        project=self.name,
        name=name,
        kind=kind,
        uuid=uuid,
        description=description,
        labels=labels,
        embedded=embedded,
        **kwargs,
    )
    self.refresh()
    return obj

new_secret(name, uuid=None, description=None, labels=None, embedded=True, secret_value=None, **kwargs)

Create a new object.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
uuid	str	ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).	None
description	str	Description of the object (human readable).	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object spec must be embedded in project spec.	True
secret_value	str	Value of the secret.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Secret	Object instance.

Examples:

>>> obj = project.new_secret(name="my-secret",
>>>                          secret_value="my-secret-value")

Source code in digitalhub_core/entities/project/entity/core.py

def new_secret(
    self,
    name: str,
    uuid: str | None = None,
    description: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    secret_value: str | None = None,
    **kwargs,
) -> Secret:
    """
    Create a new object.

    Parameters
    ----------
    name : str
        Object name.
    uuid : str
        ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
    description : str
        Description of the object (human readable).
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object spec must be embedded in project spec.
    secret_value : str
        Value of the secret.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Secret
        Object instance.

    Examples
    --------
    >>> obj = project.new_secret(name="my-secret",
    >>>                          secret_value="my-secret-value")
    """
    obj = new_secret(
        project=self.name,
        name=name,
        uuid=uuid,
        description=description,
        labels=labels,
        embedded=embedded,
        secret_value=secret_value,
        **kwargs,
    )
    self.refresh()
    return obj

new_workflow(name, kind, uuid=None, description=None, labels=None, embedded=True, **kwargs)

Create a new object.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
uuid	str	ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).	None
description	str	Description of the object (human readable).	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object spec must be embedded in project spec.	True
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Workflow	Object instance.

Examples:

>>> obj = project.new_workflow(name="my-workflow",
>>>                            kind="kfp",
>>>                            code_src="pipeline.py",
>>>                            handler="pipeline-handler")

Source code in digitalhub_core/entities/project/entity/core.py

def new_workflow(
    self,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    **kwargs,
) -> Workflow:
    """
    Create a new object.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
    description : str
        Description of the object (human readable).
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object spec must be embedded in project spec.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Workflow
        Object instance.

    Examples
    --------
    >>> obj = project.new_workflow(name="my-workflow",
    >>>                            kind="kfp",
    >>>                            code_src="pipeline.py",
    >>>                            handler="pipeline-handler")
    """
    obj = new_workflow(
        project=self.name,
        name=name,
        kind=kind,
        uuid=uuid,
        description=description,
        labels=labels,
        embedded=embedded,
        **kwargs,
    )
    self.refresh()
    return obj

update_artifact(entity)

Update object. Note that object spec are immutable.

Parameters:

Name	Type	Description	Default
entity	Artifact	Object to update.	required

Returns:

Type	Description
Artifact	Entity updated.

Examples:

>>> obj = project.update_artifact(obj)

Source code in digitalhub_core/entities/project/entity/core.py

def update_artifact(self, entity: Artifact) -> Artifact:
    """
    Update object. Note that object spec are immutable.

    Parameters
    ----------
    entity : Artifact
        Object to update.

    Returns
    -------
    Artifact
        Entity updated.

    Examples
    --------
    >>> obj = project.update_artifact(obj)
    """
    if entity.project != self.name:
        raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
    return update_artifact(entity)

update_function(entity)

Update object. Note that object spec are immutable.

Parameters:

Name	Type	Description	Default
entity	Function	Object to update.	required

Returns:

Type	Description
Function	Entity updated.

Examples:

>>> obj = project.update_function(obj)

Source code in digitalhub_core/entities/project/entity/core.py

def update_function(self, entity: Function) -> Function:
    """
    Update object. Note that object spec are immutable.

    Parameters
    ----------
    entity : Function
        Object to update.

    Returns
    -------
    Function
        Entity updated.

    Examples
    --------
    >>> obj = project.update_function(obj)
    """
    if entity.project != self.name:
        raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
    return update_function(entity)

update_secret(entity)

Update object. Note that object spec are immutable.

Parameters:

Name	Type	Description	Default
entity	Secret	Object to update.	required

Returns:

Type	Description
Secret	Entity updated.

Examples:

>>> obj = project.update_secret(obj)

Source code in digitalhub_core/entities/project/entity/core.py

def update_secret(self, entity: Secret) -> Secret:
    """
    Update object. Note that object spec are immutable.

    Parameters
    ----------
    entity : Secret
        Object to update.

    Returns
    -------
    Secret
        Entity updated.

    Examples
    --------
    >>> obj = project.update_secret(obj)
    """
    if entity.project != self.name:
        raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
    return update_secret(entity)

update_workflow(entity)

Update object. Note that object spec are immutable.

Parameters:

Name	Type	Description	Default
entity	Workflow	Object to update.	required

Returns:

Type	Description
Workflow	Entity updated.

Examples:

>>> obj = project.update_workflow(obj)

Source code in digitalhub_core/entities/project/entity/core.py

def update_workflow(self, entity: Workflow) -> Workflow:
    """
    Update object. Note that object spec are immutable.

    Parameters
    ----------
    entity : Workflow
        Object to update.

    Returns
    -------
    Workflow
        Entity updated.

    Examples
    --------
    >>> obj = project.update_workflow(obj)
    """
    if entity.project != self.name:
        raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
    return update_workflow(entity)

ProjectData

Bases: ProjectCore

ProjectData class.

Source code in digitalhub_data/entities/project/entity/data.py

class ProjectData(ProjectCore):

    """
    ProjectData class.
    """

    ##############################
    #  Dataitems
    ##############################

    def new_dataitem(
        self,
        name: str,
        kind: str,
        uuid: str | None = None,
        description: str | None = None,
        labels: list[str] | None = None,
        embedded: bool = True,
        path: str | None = None,
        **kwargs,
    ) -> Dataitem:
        """
        Create a new object.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        uuid : str
            ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
        description : str
            Description of the object (human readable).
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object spec must be embedded in project spec.
        path : str
            Object path on local file system or remote storage. It is also the destination path of upload() method.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Dataitem
            Object instance.

        Examples
        --------
        >>> obj = project.new_dataitem(name="my-dataitem",
        >>>                            kind="dataitem",
        >>>                            path="s3://my-bucket/my-key")
        """
        obj = new_dataitem(
            project=self.name,
            name=name,
            kind=kind,
            uuid=uuid,
            description=description,
            labels=labels,
            embedded=embedded,
            path=path,
            **kwargs,
        )
        self.refresh()
        return obj

    def log_dataitem(
        self,
        name: str,
        kind: str,
        source: str | None = None,
        data: Any | None = None,
        extension: str | None = None,
        path: str | None = None,
        **kwargs,
    ) -> Dataitem:
        """
        Create and upload an object.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        data : Any
            Dataframe to log.
        extension : str
            Extension of the dataitem.
        source : str
            Dataitem location on local path.
        data : Any
            Dataframe to log. Alternative to source.
        extension : str
            Extension of the output dataframe.
        path : str
            Destination path of the dataitem. If not provided, it's generated.
        **kwargs : dict
            New dataitem spec parameters.

        Returns
        -------
        Dataitem
            Object instance.

        Examples
        --------
        >>> obj = project.log_dataitem(name="my-dataitem",
        >>>                            kind="table",
        >>>                            data=df)
        """
        obj = log_dataitem(
            project=self.name,
            name=name,
            kind=kind,
            path=path,
            source=source,
            data=data,
            extension=extension,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_dataitem(
        self,
        identifier: str,
        entity_id: str | None = None,
        **kwargs,
    ) -> Dataitem:
        """
        Get object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Dataitem
            Object instance.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_dataitem("store://my-dataitem-key")

        Using entity name:
        >>> obj = project.get_dataitem("my-dataitem-name"
        >>>                            entity_id="my-dataitem-id")
        """
        obj = get_dataitem(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_dataitem_versions(
        self,
        identifier: str,
        **kwargs,
    ) -> list[Dataitem]:
        """
        Get object versions from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Dataitem]
            List of object instances.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_dataitem_versions("store://my-dataitem-key")

        Using entity name:
        >>> obj = project.get_dataitem_versions("my-dataitem-name")
        """
        return get_dataitem_versions(identifier, project=self.name, **kwargs)

    def list_dataitems(self, **kwargs) -> list[Dataitem]:
        """
        List all latest version objects from backend.

        Parameters
        ----------
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Dataitem]
            List of object instances.

        Examples
        --------
        >>> objs = project.list_dataitems()
        """
        return list_dataitems(self.name, **kwargs)

    def import_dataitem(
        self,
        file: str,
        **kwargs,
    ) -> Dataitem:
        """
        Import object from a YAML file.

        Parameters
        ----------
        file : str
            Path to YAML file.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Dataitem
            Object instance.

        Examples
        --------
        >>> obj = project.import_dataitem("my-dataitem.yaml")
        """
        return import_dataitem(file, **kwargs)

    def update_dataitem(self, entity: Dataitem) -> Dataitem:
        """
        Update object. Note that object spec are immutable.

        Parameters
        ----------
        entity : Dataitem
            Object to update.

        Returns
        -------
        Dataitem
            Entity updated.

        Examples
        --------
        >>> obj = project.update_dataitem(obj)
        """
        if entity.project != self.name:
            raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
        return update_dataitem(entity)

    def delete_dataitem(
        self,
        identifier: str,
        entity_id: str | None = None,
        delete_all_versions: bool = False,
        **kwargs,
    ) -> None:
        """
        Delete object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        dict
            Response from backend.

        Examples
        --------
        If delete_all_versions is False:
        >>> project.delete_dataitem("store://my-dataitem-key")

        Otherwise:
        >>> project.delete_dataitem("my-dataitem-name",
        >>>                         project="my-project",
        >>>                         delete_all_versions=True)
        """
        delete_dataitem(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            **kwargs,
        )
        self.refresh()

delete_dataitem(identifier, entity_id=None, delete_all_versions=False, **kwargs)

Delete object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.	False
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
dict	Response from backend.

Examples:

If delete_all_versions is False:

>>> project.delete_dataitem("store://my-dataitem-key")

Otherwise:

>>> project.delete_dataitem("my-dataitem-name",
>>>                         project="my-project",
>>>                         delete_all_versions=True)

Source code in digitalhub_data/entities/project/entity/data.py

def delete_dataitem(
    self,
    identifier: str,
    entity_id: str | None = None,
    delete_all_versions: bool = False,
    **kwargs,
) -> None:
    """
    Delete object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    dict
        Response from backend.

    Examples
    --------
    If delete_all_versions is False:
    >>> project.delete_dataitem("store://my-dataitem-key")

    Otherwise:
    >>> project.delete_dataitem("my-dataitem-name",
    >>>                         project="my-project",
    >>>                         delete_all_versions=True)
    """
    delete_dataitem(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        delete_all_versions=delete_all_versions,
        **kwargs,
    )
    self.refresh()

get_dataitem(identifier, entity_id=None, **kwargs)

Get object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Dataitem	Object instance.

Examples:

Using entity key:

>>> obj = project.get_dataitem("store://my-dataitem-key")

Using entity name:

>>> obj = project.get_dataitem("my-dataitem-name"
>>>                            entity_id="my-dataitem-id")

Source code in digitalhub_data/entities/project/entity/data.py

def get_dataitem(
    self,
    identifier: str,
    entity_id: str | None = None,
    **kwargs,
) -> Dataitem:
    """
    Get object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Dataitem
        Object instance.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_dataitem("store://my-dataitem-key")

    Using entity name:
    >>> obj = project.get_dataitem("my-dataitem-name"
    >>>                            entity_id="my-dataitem-id")
    """
    obj = get_dataitem(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

get_dataitem_versions(identifier, **kwargs)

Get object versions from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Dataitem]	List of object instances.

Examples:

Using entity key:

>>> obj = project.get_dataitem_versions("store://my-dataitem-key")

Using entity name:

>>> obj = project.get_dataitem_versions("my-dataitem-name")

Source code in digitalhub_data/entities/project/entity/data.py

def get_dataitem_versions(
    self,
    identifier: str,
    **kwargs,
) -> list[Dataitem]:
    """
    Get object versions from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Dataitem]
        List of object instances.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_dataitem_versions("store://my-dataitem-key")

    Using entity name:
    >>> obj = project.get_dataitem_versions("my-dataitem-name")
    """
    return get_dataitem_versions(identifier, project=self.name, **kwargs)

import_dataitem(file, **kwargs)

Import object from a YAML file.

Parameters:

Name	Type	Description	Default
file	str	Path to YAML file.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Dataitem	Object instance.

Examples:

>>> obj = project.import_dataitem("my-dataitem.yaml")

Source code in digitalhub_data/entities/project/entity/data.py

def import_dataitem(
    self,
    file: str,
    **kwargs,
) -> Dataitem:
    """
    Import object from a YAML file.

    Parameters
    ----------
    file : str
        Path to YAML file.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Dataitem
        Object instance.

    Examples
    --------
    >>> obj = project.import_dataitem("my-dataitem.yaml")
    """
    return import_dataitem(file, **kwargs)

list_dataitems(**kwargs)

List all latest version objects from backend.

Parameters:

Name	Type	Description	Default
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Dataitem]	List of object instances.

Examples:

>>> objs = project.list_dataitems()

Source code in digitalhub_data/entities/project/entity/data.py

def list_dataitems(self, **kwargs) -> list[Dataitem]:
    """
    List all latest version objects from backend.

    Parameters
    ----------
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Dataitem]
        List of object instances.

    Examples
    --------
    >>> objs = project.list_dataitems()
    """
    return list_dataitems(self.name, **kwargs)

log_dataitem(name, kind, source=None, data=None, extension=None, path=None, **kwargs)

Create and upload an object.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
data	Any	Dataframe to log.	None
extension	str	Extension of the dataitem.	None
source	str	Dataitem location on local path.	None
data	Any	Dataframe to log. Alternative to source.	None
extension	str	Extension of the output dataframe.	None
path	str	Destination path of the dataitem. If not provided, it's generated.	None
**kwargs	dict	New dataitem spec parameters.	{}

Returns:

Type	Description
Dataitem	Object instance.

Examples:

>>> obj = project.log_dataitem(name="my-dataitem",
>>>                            kind="table",
>>>                            data=df)

Source code in digitalhub_data/entities/project/entity/data.py

def log_dataitem(
    self,
    name: str,
    kind: str,
    source: str | None = None,
    data: Any | None = None,
    extension: str | None = None,
    path: str | None = None,
    **kwargs,
) -> Dataitem:
    """
    Create and upload an object.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    data : Any
        Dataframe to log.
    extension : str
        Extension of the dataitem.
    source : str
        Dataitem location on local path.
    data : Any
        Dataframe to log. Alternative to source.
    extension : str
        Extension of the output dataframe.
    path : str
        Destination path of the dataitem. If not provided, it's generated.
    **kwargs : dict
        New dataitem spec parameters.

    Returns
    -------
    Dataitem
        Object instance.

    Examples
    --------
    >>> obj = project.log_dataitem(name="my-dataitem",
    >>>                            kind="table",
    >>>                            data=df)
    """
    obj = log_dataitem(
        project=self.name,
        name=name,
        kind=kind,
        path=path,
        source=source,
        data=data,
        extension=extension,
        **kwargs,
    )
    self.refresh()
    return obj

new_dataitem(name, kind, uuid=None, description=None, labels=None, embedded=True, path=None, **kwargs)

Create a new object.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
uuid	str	ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).	None
description	str	Description of the object (human readable).	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object spec must be embedded in project spec.	True
path	str	Object path on local file system or remote storage. It is also the destination path of upload() method.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Dataitem	Object instance.

Examples:

>>> obj = project.new_dataitem(name="my-dataitem",
>>>                            kind="dataitem",
>>>                            path="s3://my-bucket/my-key")

Source code in digitalhub_data/entities/project/entity/data.py

def new_dataitem(
    self,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    path: str | None = None,
    **kwargs,
) -> Dataitem:
    """
    Create a new object.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
    description : str
        Description of the object (human readable).
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object spec must be embedded in project spec.
    path : str
        Object path on local file system or remote storage. It is also the destination path of upload() method.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Dataitem
        Object instance.

    Examples
    --------
    >>> obj = project.new_dataitem(name="my-dataitem",
    >>>                            kind="dataitem",
    >>>                            path="s3://my-bucket/my-key")
    """
    obj = new_dataitem(
        project=self.name,
        name=name,
        kind=kind,
        uuid=uuid,
        description=description,
        labels=labels,
        embedded=embedded,
        path=path,
        **kwargs,
    )
    self.refresh()
    return obj

update_dataitem(entity)

Update object. Note that object spec are immutable.

Parameters:

Name	Type	Description	Default
entity	Dataitem	Object to update.	required

Returns:

Type	Description
Dataitem	Entity updated.

Examples:

>>> obj = project.update_dataitem(obj)

Source code in digitalhub_data/entities/project/entity/data.py

def update_dataitem(self, entity: Dataitem) -> Dataitem:
    """
    Update object. Note that object spec are immutable.

    Parameters
    ----------
    entity : Dataitem
        Object to update.

    Returns
    -------
    Dataitem
        Entity updated.

    Examples
    --------
    >>> obj = project.update_dataitem(obj)
    """
    if entity.project != self.name:
        raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
    return update_dataitem(entity)

ProjectMl

Bases: ProjectData

ProjectMl class.

Source code in digitalhub_ml/entities/project/entity/ml.py

class ProjectMl(ProjectData):
    """
    ProjectMl class.
    """

    ##############################
    #  Models
    ##############################

    def new_model(
        self,
        name: str,
        kind: str,
        uuid: str | None = None,
        description: str | None = None,
        labels: list[str] | None = None,
        embedded: bool = True,
        path: str | None = None,
        **kwargs,
    ) -> Model:
        """
        Create a new object.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        uuid : str
            ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
        description : str
            Description of the object (human readable).
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object spec must be embedded in project spec.
        path : str
            Object path on local file system or remote storage. It is also the destination path of upload() method.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Model
            Object instance.

        Examples
        --------
        >>> obj = project.new_model(name="my-model",
        >>>                            kind="model",
        >>>                            path="s3://my-bucket/my-key")
        """
        obj = new_model(
            project=self.name,
            name=name,
            kind=kind,
            uuid=uuid,
            description=description,
            labels=labels,
            embedded=embedded,
            path=path,
            **kwargs,
        )
        self.refresh()
        return obj

    def log_model(
        self,
        name: str,
        kind: str,
        source: str,
        path: str | None = None,
        **kwargs,
    ) -> Model:
        """
        Create and upload an object.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        source : str
            Model location on local path.
        path : str
            Destination path of the model. If not provided, it's generated.
        **kwargs : dict
            New model spec parameters.

        Returns
        -------
        Model
            Object instance.

        Examples
        --------
        >>> obj = project.log_model(name="my-model",
        >>>                            kind="model",
        >>>                            source="./local-path")
        """
        obj = log_model(
            project=self.name,
            name=name,
            kind=kind,
            source=source,
            path=path,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_model(
        self,
        identifier: str,
        entity_id: str | None = None,
        **kwargs,
    ) -> Model:
        """
        Get object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Model
            Object instance.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_model("store://my-model-key")

        Using entity name:
        >>> obj = project.get_model("my-model-name"
        >>>                            entity_id="my-model-id")
        """
        obj = get_model(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

    def get_model_versions(
        self,
        identifier: str,
        **kwargs,
    ) -> list[Model]:
        """
        Get object versions from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Model]
            List of object instances.

        Examples
        --------
        Using entity key:
        >>> obj = project.get_model_versions("store://my-model-key")

        Using entity name:
        >>> obj = project.get_model_versions("my-model-name")
        """
        return get_model_versions(identifier, project=self.name, **kwargs)

    def list_models(self, **kwargs) -> list[Model]:
        """
        List all latest version objects from backend.

        Parameters
        ----------
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        list[Model]
            List of object instances.

        Examples
        --------
        >>> objs = project.list_models()
        """
        return list_models(self.name, **kwargs)

    def import_model(
        self,
        file: str,
        **kwargs,
    ) -> Model:
        """
        Import object from a YAML file.

        Parameters
        ----------
        file : str
            Path to YAML file.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Model
            Object instance.

        Examples
        --------
        >>> obj = project.import_model("my-model.yaml")
        """
        return import_model(file, **kwargs)

    def update_model(self, entity: Model) -> Model:
        """
        Update object. Note that object spec are immutable.

        Parameters
        ----------
        entity : Model
            Object to update.

        Returns
        -------
        Model
            Entity updated.

        Examples
        --------
        >>> obj = project.update_model(obj)
        """
        if entity.project != self.name:
            raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
        return update_model(entity)

    def delete_model(
        self,
        identifier: str,
        entity_id: str | None = None,
        delete_all_versions: bool = False,
        **kwargs,
    ) -> None:
        """
        Delete object from backend.

        Parameters
        ----------
        identifier : str
            Entity key (store://...) or entity name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        dict
            Response from backend.

        Examples
        --------
        If delete_all_versions is False:
        >>> project.delete_model("store://my-model-key")

        Otherwise:
        >>> project.delete_model("my-model-name",
        >>>                         project="my-project",
        >>>                         delete_all_versions=True)
        """
        delete_model(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            **kwargs,
        )
        self.refresh()

delete_model(identifier, entity_id=None, delete_all_versions=False, **kwargs)

Delete object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.	False
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
dict	Response from backend.

Examples:

If delete_all_versions is False:

>>> project.delete_model("store://my-model-key")

Otherwise:

>>> project.delete_model("my-model-name",
>>>                         project="my-project",
>>>                         delete_all_versions=True)

Source code in digitalhub_ml/entities/project/entity/ml.py

def delete_model(
    self,
    identifier: str,
    entity_id: str | None = None,
    delete_all_versions: bool = False,
    **kwargs,
) -> None:
    """
    Delete object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity. If True, use entity name instead of entity key as identifier.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    dict
        Response from backend.

    Examples
    --------
    If delete_all_versions is False:
    >>> project.delete_model("store://my-model-key")

    Otherwise:
    >>> project.delete_model("my-model-name",
    >>>                         project="my-project",
    >>>                         delete_all_versions=True)
    """
    delete_model(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        delete_all_versions=delete_all_versions,
        **kwargs,
    )
    self.refresh()

get_model(identifier, entity_id=None, **kwargs)

Get object from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
entity_id	str	Entity ID.	None
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Model	Object instance.

Examples:

Using entity key:

>>> obj = project.get_model("store://my-model-key")

Using entity name:

>>> obj = project.get_model("my-model-name"
>>>                            entity_id="my-model-id")

Source code in digitalhub_ml/entities/project/entity/ml.py

def get_model(
    self,
    identifier: str,
    entity_id: str | None = None,
    **kwargs,
) -> Model:
    """
    Get object from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    entity_id : str
        Entity ID.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Model
        Object instance.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_model("store://my-model-key")

    Using entity name:
    >>> obj = project.get_model("my-model-name"
    >>>                            entity_id="my-model-id")
    """
    obj = get_model(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

get_model_versions(identifier, **kwargs)

Get object versions from backend.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key (store://...) or entity name.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Model]	List of object instances.

Examples:

Using entity key:

>>> obj = project.get_model_versions("store://my-model-key")

Using entity name:

>>> obj = project.get_model_versions("my-model-name")

Source code in digitalhub_ml/entities/project/entity/ml.py

def get_model_versions(
    self,
    identifier: str,
    **kwargs,
) -> list[Model]:
    """
    Get object versions from backend.

    Parameters
    ----------
    identifier : str
        Entity key (store://...) or entity name.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Model]
        List of object instances.

    Examples
    --------
    Using entity key:
    >>> obj = project.get_model_versions("store://my-model-key")

    Using entity name:
    >>> obj = project.get_model_versions("my-model-name")
    """
    return get_model_versions(identifier, project=self.name, **kwargs)

import_model(file, **kwargs)

Import object from a YAML file.

Parameters:

Name	Type	Description	Default
file	str	Path to YAML file.	required
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Model	Object instance.

Examples:

>>> obj = project.import_model("my-model.yaml")

Source code in digitalhub_ml/entities/project/entity/ml.py

def import_model(
    self,
    file: str,
    **kwargs,
) -> Model:
    """
    Import object from a YAML file.

    Parameters
    ----------
    file : str
        Path to YAML file.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    Model
        Object instance.

    Examples
    --------
    >>> obj = project.import_model("my-model.yaml")
    """
    return import_model(file, **kwargs)

list_models(**kwargs)

List all latest version objects from backend.

Parameters:

Name	Type	Description	Default
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
list[Model]	List of object instances.

Examples:

>>> objs = project.list_models()

Source code in digitalhub_ml/entities/project/entity/ml.py

def list_models(self, **kwargs) -> list[Model]:
    """
    List all latest version objects from backend.

    Parameters
    ----------
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    list[Model]
        List of object instances.

    Examples
    --------
    >>> objs = project.list_models()
    """
    return list_models(self.name, **kwargs)

log_model(name, kind, source, path=None, **kwargs)

Create and upload an object.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
source	str	Model location on local path.	required
path	str	Destination path of the model. If not provided, it's generated.	None
**kwargs	dict	New model spec parameters.	{}

Returns:

Type	Description
Model	Object instance.

Examples:

>>> obj = project.log_model(name="my-model",
>>>                            kind="model",
>>>                            source="./local-path")

Source code in digitalhub_ml/entities/project/entity/ml.py

def log_model(
    self,
    name: str,
    kind: str,
    source: str,
    path: str | None = None,
    **kwargs,
) -> Model:
    """
    Create and upload an object.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    source : str
        Model location on local path.
    path : str
        Destination path of the model. If not provided, it's generated.
    **kwargs : dict
        New model spec parameters.

    Returns
    -------
    Model
        Object instance.

    Examples
    --------
    >>> obj = project.log_model(name="my-model",
    >>>                            kind="model",
    >>>                            source="./local-path")
    """
    obj = log_model(
        project=self.name,
        name=name,
        kind=kind,
        source=source,
        path=path,
        **kwargs,
    )
    self.refresh()
    return obj

new_model(name, kind, uuid=None, description=None, labels=None, embedded=True, path=None, **kwargs)

Create a new object.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
uuid	str	ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).	None
description	str	Description of the object (human readable).	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object spec must be embedded in project spec.	True
path	str	Object path on local file system or remote storage. It is also the destination path of upload() method.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Model	Object instance.

Examples:

>>> obj = project.new_model(name="my-model",
>>>                            kind="model",
>>>                            path="s3://my-bucket/my-key")

Source code in digitalhub_ml/entities/project/entity/ml.py

def new_model(
    self,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    path: str | None = None,
    **kwargs,
) -> Model:
    """
    Create a new object.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4, e.g. 40f25c4b-d26b-4221-b048-9527aff291e2).
    description : str
        Description of the object (human readable).
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object spec must be embedded in project spec.
    path : str
        Object path on local file system or remote storage. It is also the destination path of upload() method.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Model
        Object instance.

    Examples
    --------
    >>> obj = project.new_model(name="my-model",
    >>>                            kind="model",
    >>>                            path="s3://my-bucket/my-key")
    """
    obj = new_model(
        project=self.name,
        name=name,
        kind=kind,
        uuid=uuid,
        description=description,
        labels=labels,
        embedded=embedded,
        path=path,
        **kwargs,
    )
    self.refresh()
    return obj

update_model(entity)

Update object. Note that object spec are immutable.

Parameters:

Name	Type	Description	Default
entity	Model	Object to update.	required

Returns:

Type	Description
Model	Entity updated.

Examples:

>>> obj = project.update_model(obj)

Source code in digitalhub_ml/entities/project/entity/ml.py

def update_model(self, entity: Model) -> Model:
    """
    Update object. Note that object spec are immutable.

    Parameters
    ----------
    entity : Model
        Object to update.

    Returns
    -------
    Model
        Entity updated.

    Examples
    --------
    >>> obj = project.update_model(obj)
    """
    if entity.project != self.name:
        raise ValueError(f"Entity {entity.name} is not in project {self.name}.")
    return update_model(entity)