Entity and methods

Project

Bases: Entity

A class representing a project.

Source code in digitalhub_core/entities/projects/entity.py

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

    ENTITY_TYPE = EntityTypes.PROJECTS.value

    def __init__(
        self,
        name: str,
        kind: str,
        metadata: Metadata,
        spec: ProjectSpec,
        status: ProjectStatus,
        user: str | None = None,
        local: bool = False,
    ) -> None:
        """
        Constructor.

        Parameters
        ----------
        name : str
            Name of the object.
        kind : str
            Kind the object.
        metadata : Metadata
            Metadata of the object.
        spec : ProjectSpec
            Specification of the object.
        status : ProjectStatus
            Status of the object.
        user : str
            Owner of the object.
        local: bool
            If True, export locally.
        """
        super().__init__()
        self.id = name
        self.name = name
        self.kind = kind
        self.key = f"store://{name}"
        self.metadata = metadata
        self.spec = spec
        self.status = status
        self.user = user

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

        # 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()

    #############################
    #  Artifacts
    #############################

    def new_artifact(
        self,
        name: str,
        kind: str,
        uuid: str | None = None,
        description: str | None = None,
        git_source: str | None = None,
        labels: list[str] | None = None,
        embedded: bool = True,
        path: str | None = None,
        src_path: str | None = None,
        **kwargs,
    ) -> Artifact:
        """
        Create an instance of the Artifact class with the provided parameters.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        uuid : str
            ID of the object (UUID4).
        description : str
            Description of the object (human readable).
        git_source : str
            Remote git source for object.
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object must be embedded in project.
        path : str
            Object path on local file system or remote storage.
            If not provided, it's generated.
        src_path : str
            Local object path.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Artifact
            Object instance.
        """
        obj = new_artifact(
            project=self.name,
            name=name,
            kind=kind,
            uuid=uuid,
            description=description,
            git_source=git_source,
            labels=labels,
            embedded=embedded,
            path=path,
            src_path=src_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 or name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Artifact
            Instance of Artifact class.
        """
        obj = get_artifact(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

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

        Parameters
        ----------
        identifier : str
            Entity key or name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity.
            Use entity name instead of entity key as identifier.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        None
        """
        delete_artifact(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            **kwargs,
        )
        self.refresh()

    def list_artifacts(self, **kwargs) -> list[dict]:
        """
        List artifacts associated with the project.

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

        Returns
        -------
        list[dict]
            List of objects related to project.
        """
        return list_artifacts(self.name, **kwargs)

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

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        path : str
            Destination path of the artifact.
        source_path : str
            Artifact location on local machine.
        **kwargs : dict
            New artifact parameters.

        Returns
        -------
        Artifact
            Instance of Artifact class.
        """
        obj = log_artifact(
            project=self.name,
            name=name,
            kind=kind,
            path=path,
            source_path=source_path,
            **kwargs,
        )
        self.refresh()
        return obj

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

    def new_function(
        self,
        name: str,
        kind: str,
        uuid: str | None = None,
        description: str | None = None,
        git_source: str | None = None,
        labels: list[str] | None = None,
        embedded: bool = True,
        **kwargs,
    ) -> Function:
        """
        Create a Function instance with the given parameters.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        uuid : str
            ID of the object (UUID4).
        description : str
            Description of the object (human readable).
        git_source : str
            Remote git source for object.
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object must be embedded in project.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Function
            Object instance.
        """
        obj = new_function(
            project=self.name,
            name=name,
            kind=kind,
            uuid=uuid,
            description=description,
            git_source=git_source,
            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 or name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Function
            Instance of Function class.
        """
        obj = get_function(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

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

        Parameters
        ----------
        identifier : str
            Entity key or name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity.
            Use entity name instead of entity key as identifier.
        cascade : bool
            Cascade delete.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        None
        """
        delete_function(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            cascade=cascade,
            **kwargs,
        )
        self.refresh()

    def list_functions(self, **kwargs) -> list[dict]:
        """
        List functions associated with the project.

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

        Returns
        -------
        list[dict]
            List of objects related to project.
        """
        return list_functions(self.name, **kwargs)

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

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

        Parameters
        ----------
        name : str
            Object name.
        uuid : str
            ID of the object (UUID4).
        description : str
            Description of the object (human readable).
        git_source : str
            Remote git source for object.
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object must be embedded in project.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Workflow
            An instance of the created workflow.
        """
        obj = new_workflow(
            project=self.name,
            name=name,
            kind=kind,
            uuid=uuid,
            description=description,
            git_source=git_source,
            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 or name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Workflow
            Instance of Workflow class.
        """
        obj = get_workflow(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

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

        Parameters
        ----------
        identifier : str
            Entity key or name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity.
            Use entity name instead of entity key as identifier.
        cascade : bool
            Cascade delete.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        None
        """
        delete_workflow(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            cascade=cascade,
            **kwargs,
        )
        self.refresh()

    def list_workflows(self, **kwargs) -> list[dict]:
        """
        List workflows associated with the project.

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

        Returns
        -------
        list[dict]
            List of objects related to project.
        """
        return list_workflows(self.name, **kwargs)

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

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

        Parameters
        ----------
        name : str
            Object name.
        uuid : str
            ID of the object (UUID4).
        description : str
            Description of the object (human readable).
        git_source : str
            Remote git source for object.
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object must be embedded in project.
        secret_value : str
            Value of the secret.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Secret
            An instance of the created secret.
        """
        obj = new_secret(
            project=self.name,
            name=name,
            uuid=uuid,
            description=description,
            git_source=git_source,
            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 or name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Secret
            Instance of Secret class.
        """
        obj = get_secret(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

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

        Parameters
        ----------
        identifier : str
            Entity key or name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity.
            Use entity name instead of entity key as identifier.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        None
        """
        delete_secret(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            **kwargs,
        )
        self.refresh()

    def list_secrets(self, **kwargs) -> list[dict]:
        """
        List secrets associated with the project.

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

        Returns
        -------
        list[dict]
            List of objects related to project.
        """
        return list_secrets(self.name, **kwargs)

    #############################
    #  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,
        }

__init__(name, kind, metadata, spec, status, user=None, local=False)

Constructor.

Parameters:

Name	Type	Description	Default
name	str	Name of the object.	required
kind	str	Kind the object.	required
metadata	Metadata	Metadata of the object.	required
spec	ProjectSpec	Specification of the object.	required
status	ProjectStatus	Status of the object.	required
user	str	Owner of the object.	None
local	bool	If True, export locally.	False

Source code in digitalhub_core/entities/projects/entity.py

def __init__(
    self,
    name: str,
    kind: str,
    metadata: Metadata,
    spec: ProjectSpec,
    status: ProjectStatus,
    user: str | None = None,
    local: bool = False,
) -> None:
    """
    Constructor.

    Parameters
    ----------
    name : str
        Name of the object.
    kind : str
        Kind the object.
    metadata : Metadata
        Metadata of the object.
    spec : ProjectSpec
        Specification of the object.
    status : ProjectStatus
        Status of the object.
    user : str
        Owner of the object.
    local: bool
        If True, export locally.
    """
    super().__init__()
    self.id = name
    self.name = name
    self.kind = kind
    self.key = f"store://{name}"
    self.metadata = metadata
    self.spec = spec
    self.status = status
    self.user = user

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

    # Set client
    self._client = get_client(local)

    # Set context
    set_context(self)

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

Delete a Artifact from project.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key or name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. Use entity name instead of entity key as identifier.	False
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
None

Source code in digitalhub_core/entities/projects/entity.py

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

    Parameters
    ----------
    identifier : str
        Entity key or name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity.
        Use entity name instead of entity key as identifier.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    None
    """
    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=False, **kwargs)

Delete a Function from project.

Parameters:

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

Returns:

Type	Description
None

Source code in digitalhub_core/entities/projects/entity.py

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

    Parameters
    ----------
    identifier : str
        Entity key or name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity.
        Use entity name instead of entity key as identifier.
    cascade : bool
        Cascade delete.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    None
    """
    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 a Secret from project.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key or name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. Use entity name instead of entity key as identifier.	False
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
None

Source code in digitalhub_core/entities/projects/entity.py

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

    Parameters
    ----------
    identifier : str
        Entity key or name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity.
        Use entity name instead of entity key as identifier.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    None
    """
    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=False, **kwargs)

Delete a Workflow from project.

Parameters:

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

Returns:

Type	Description
None

Source code in digitalhub_core/entities/projects/entity.py

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

    Parameters
    ----------
    identifier : str
        Entity key or name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity.
        Use entity name instead of entity key as identifier.
    cascade : bool
        Cascade delete.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    None
    """
    delete_workflow(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        delete_all_versions=delete_all_versions,
        cascade=cascade,
        **kwargs,
    )
    self.refresh()

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/projects/entity.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)

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

Get object from backend.

Parameters:

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

Returns:

Type	Description
Artifact	Instance of Artifact class.

Source code in digitalhub_core/entities/projects/entity.py

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

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

    Returns
    -------
    Artifact
        Instance of Artifact class.
    """
    obj = get_artifact(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

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

Get object from backend.

Parameters:

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

Returns:

Type	Description
Function	Instance of Function class.

Source code in digitalhub_core/entities/projects/entity.py

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

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

    Returns
    -------
    Function
        Instance of Function class.
    """
    obj = get_function(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

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

Get object from backend.

Parameters:

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

Returns:

Type	Description
Secret	Instance of Secret class.

Source code in digitalhub_core/entities/projects/entity.py

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

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

    Returns
    -------
    Secret
        Instance of Secret class.
    """
    obj = get_secret(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

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

Get object from backend.

Parameters:

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

Returns:

Type	Description
Workflow	Instance of Workflow class.

Source code in digitalhub_core/entities/projects/entity.py

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

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

    Returns
    -------
    Workflow
        Instance of Workflow class.
    """
    obj = get_workflow(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

list_artifacts(**kwargs)

List artifacts associated with the project.

Parameters:

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

Returns:

Type	Description
list[dict]	List of objects related to project.

Source code in digitalhub_core/entities/projects/entity.py

def list_artifacts(self, **kwargs) -> list[dict]:
    """
    List artifacts associated with the project.

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

    Returns
    -------
    list[dict]
        List of objects related to project.
    """
    return list_artifacts(self.name, **kwargs)

list_functions(**kwargs)

List functions associated with the project.

Parameters:

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

Returns:

Type	Description
list[dict]	List of objects related to project.

Source code in digitalhub_core/entities/projects/entity.py

def list_functions(self, **kwargs) -> list[dict]:
    """
    List functions associated with the project.

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

    Returns
    -------
    list[dict]
        List of objects related to project.
    """
    return list_functions(self.name, **kwargs)

list_secrets(**kwargs)

List secrets associated with the project.

Parameters:

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

Returns:

Type	Description
list[dict]	List of objects related to project.

Source code in digitalhub_core/entities/projects/entity.py

def list_secrets(self, **kwargs) -> list[dict]:
    """
    List secrets associated with the project.

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

    Returns
    -------
    list[dict]
        List of objects related to project.
    """
    return list_secrets(self.name, **kwargs)

list_workflows(**kwargs)

List workflows associated with the project.

Parameters:

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

Returns:

Type	Description
list[dict]	List of objects related to project.

Source code in digitalhub_core/entities/projects/entity.py

def list_workflows(self, **kwargs) -> list[dict]:
    """
    List workflows associated with the project.

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

    Returns
    -------
    list[dict]
        List of objects related to project.
    """
    return list_workflows(self.name, **kwargs)

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

Create and upload an artifact.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
path	str	Destination path of the artifact.	None
source_path	str	Artifact location on local machine.	None
**kwargs	dict	New artifact parameters.	{}

Returns:

Type	Description
Artifact	Instance of Artifact class.

Source code in digitalhub_core/entities/projects/entity.py

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

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    path : str
        Destination path of the artifact.
    source_path : str
        Artifact location on local machine.
    **kwargs : dict
        New artifact parameters.

    Returns
    -------
    Artifact
        Instance of Artifact class.
    """
    obj = log_artifact(
        project=self.name,
        name=name,
        kind=kind,
        path=path,
        source_path=source_path,
        **kwargs,
    )
    self.refresh()
    return obj

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

Create an instance of the Artifact class with the provided parameters.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
uuid	str	ID of the object (UUID4).	None
description	str	Description of the object (human readable).	None
git_source	str	Remote git source for object.	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object must be embedded in project.	True
path	str	Object path on local file system or remote storage. If not provided, it's generated.	None
src_path	str	Local object path.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Artifact	Object instance.

Source code in digitalhub_core/entities/projects/entity.py

def new_artifact(
    self,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    git_source: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    path: str | None = None,
    src_path: str | None = None,
    **kwargs,
) -> Artifact:
    """
    Create an instance of the Artifact class with the provided parameters.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4).
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object must be embedded in project.
    path : str
        Object path on local file system or remote storage.
        If not provided, it's generated.
    src_path : str
        Local object path.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Artifact
        Object instance.
    """
    obj = new_artifact(
        project=self.name,
        name=name,
        kind=kind,
        uuid=uuid,
        description=description,
        git_source=git_source,
        labels=labels,
        embedded=embedded,
        path=path,
        src_path=src_path,
        **kwargs,
    )
    self.refresh()
    return obj

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

Create a Function instance with the given parameters.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
uuid	str	ID of the object (UUID4).	None
description	str	Description of the object (human readable).	None
git_source	str	Remote git source for object.	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object must be embedded in project.	True
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Function	Object instance.

Source code in digitalhub_core/entities/projects/entity.py

def new_function(
    self,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    git_source: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    **kwargs,
) -> Function:
    """
    Create a Function instance with the given parameters.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4).
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object must be embedded in project.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Function
        Object instance.
    """
    obj = new_function(
        project=self.name,
        name=name,
        kind=kind,
        uuid=uuid,
        description=description,
        git_source=git_source,
        labels=labels,
        embedded=embedded,
        **kwargs,
    )
    self.refresh()
    return obj

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

Create a new Secret instance with the specified parameters.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
uuid	str	ID of the object (UUID4).	None
description	str	Description of the object (human readable).	None
git_source	str	Remote git source for object.	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object must be embedded in project.	True
secret_value	str	Value of the secret.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Secret	An instance of the created secret.

Source code in digitalhub_core/entities/projects/entity.py

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

    Parameters
    ----------
    name : str
        Object name.
    uuid : str
        ID of the object (UUID4).
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object must be embedded in project.
    secret_value : str
        Value of the secret.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Secret
        An instance of the created secret.
    """
    obj = new_secret(
        project=self.name,
        name=name,
        uuid=uuid,
        description=description,
        git_source=git_source,
        labels=labels,
        embedded=embedded,
        secret_value=secret_value,
        **kwargs,
    )
    self.refresh()
    return obj

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

Create a new Workflow instance with the specified parameters.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
uuid	str	ID of the object (UUID4).	None
description	str	Description of the object (human readable).	None
git_source	str	Remote git source for object.	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object must be embedded in project.	True
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Workflow	An instance of the created workflow.

Source code in digitalhub_core/entities/projects/entity.py

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

    Parameters
    ----------
    name : str
        Object name.
    uuid : str
        ID of the object (UUID4).
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object must be embedded in project.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Workflow
        An instance of the created workflow.
    """
    obj = new_workflow(
        project=self.name,
        name=name,
        kind=kind,
        uuid=uuid,
        description=description,
        git_source=git_source,
        labels=labels,
        embedded=embedded,
        **kwargs,
    )
    self.refresh()
    return obj

refresh()

Refresh object from backend.

Returns:

Type	Description
Project	Project object.

Source code in digitalhub_core/entities/projects/entity.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/projects/entity.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

project_from_dict(obj)

Create project from dictionary.

Parameters:

Name	Type	Description	Default
obj	dict	Dictionary to create object from.	required

Returns:

Type	Description
Project	Project object.

Source code in digitalhub_core/entities/projects/entity.py

def project_from_dict(obj: dict) -> Project:
    """
    Create project from dictionary.

    Parameters
    ----------
    obj : dict
        Dictionary to create object from.

    Returns
    -------
    Project
        Project object.
    """
    return Project.from_dict(obj)

project_from_parameters(name, kind, description=None, git_source=None, labels=None, local=False, context=None, **kwargs)

Create project.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
description	str	Description of the object (human readable).	None
git_source	str	Remote git source for object.	None
labels	list[str]	List of labels.	None
local	bool	Flag to determine if object will be exported to backend.	False
context	str	The context of the project.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Project	Project object.

Source code in digitalhub_core/entities/projects/entity.py

def project_from_parameters(
    name: str,
    kind: str,
    description: str | None = None,
    git_source: str | None = None,
    labels: list[str] | None = None,
    local: bool = False,
    context: str | None = None,
    **kwargs,
) -> Project:
    """
    Create project.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    local : bool
        Flag to determine if object will be exported to backend.
    context : str
        The context of the project.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Project
        Project object.
    """
    name = build_name(name)
    spec = build_spec(
        kind,
        context=context,
        **kwargs,
    )
    metadata = build_metadata(
        kind,
        project=name,
        name=name,
        description=description,
        labels=labels,
        source=git_source,
    )
    status = build_status(kind)
    return Project(
        name=name,
        kind=kind,
        metadata=metadata,
        spec=spec,
        status=status,
        local=local,
    )

ProjectData

Bases: Project

ProjectData class.

Source code in digitalhub_data/entities/projects/entity.py

class ProjectData(Project):

    """
    ProjectData class.
    """

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

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

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        uuid : str
            ID of the object (UUID4).
        description : str
            Description of the object (human readable).
        git_source : str
            Remote git source for object.
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object must be embedded in project.
        path : str
            Object path on local file system or remote storage.
            If not provided, it's generated.
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Dataitem
            Object instance.
        """
        obj = new_dataitem(
            project=self.name,
            name=name,
            kind=kind,
            path=path,
            uuid=uuid,
            description=description,
            git_source=git_source,
            labels=labels,
            embedded=embedded,
            **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 or name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Dataitem
            Instance of Dataitem class.
        """
        obj = get_dataitem(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

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

        Parameters
        ----------
        identifier : str
            Entity key or name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity.
            Use entity name instead of entity key as identifier.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        None
        """
        delete_dataitem(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            **kwargs,
        )
        self.refresh()

    def list_dataitems(self, **kwargs) -> list[dict]:
        """
        List dataitems associated with the project.

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

        Returns
        -------
        list[dict]
            List of objects related to project.
        """
        return list_dataitems(self.name, **kwargs)

    def log_dataitem(
        self,
        name: str,
        kind: str,
        path: str | None = None,
        data: Any | None = None,
        extension: str | None = None,
        **kwargs,
    ) -> Dataitem:
        """
        Log a dataitem to the project.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        path : str
            Destination path of the dataitem.
        data : Any
            Dataframe to log.
        extension : str
            Extension of the dataitem.
        **kwargs : dict
            New dataitem parameters.

        Returns
        -------
        Dataitem
            Object instance.
        """
        obj = log_dataitem(
            project=self.name,
            name=name,
            kind=kind,
            path=path,
            data=data,
            extension=extension,
            **kwargs,
        )
        self.refresh()
        return obj

    @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.
        """
        # Override methods to search in digitalhub_data
        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,
        }

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

Delete a Dataitem from project.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key or name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. Use entity name instead of entity key as identifier.	False
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
None

Source code in digitalhub_data/entities/projects/entity.py

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

    Parameters
    ----------
    identifier : str
        Entity key or name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity.
        Use entity name instead of entity key as identifier.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    None
    """
    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 or name.	required
entity_id	str	Entity ID.	None
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Dataitem	Instance of Dataitem class.

Source code in digitalhub_data/entities/projects/entity.py

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

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

    Returns
    -------
    Dataitem
        Instance of Dataitem class.
    """
    obj = get_dataitem(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

list_dataitems(**kwargs)

List dataitems associated with the project.

Parameters:

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

Returns:

Type	Description
list[dict]	List of objects related to project.

Source code in digitalhub_data/entities/projects/entity.py

def list_dataitems(self, **kwargs) -> list[dict]:
    """
    List dataitems associated with the project.

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

    Returns
    -------
    list[dict]
        List of objects related to project.
    """
    return list_dataitems(self.name, **kwargs)

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

Log a dataitem to the project.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
path	str	Destination path of the dataitem.	None
data	Any	Dataframe to log.	None
extension	str	Extension of the dataitem.	None
**kwargs	dict	New dataitem parameters.	{}

Returns:

Type	Description
Dataitem	Object instance.

Source code in digitalhub_data/entities/projects/entity.py

def log_dataitem(
    self,
    name: str,
    kind: str,
    path: str | None = None,
    data: Any | None = None,
    extension: str | None = None,
    **kwargs,
) -> Dataitem:
    """
    Log a dataitem to the project.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    path : str
        Destination path of the dataitem.
    data : Any
        Dataframe to log.
    extension : str
        Extension of the dataitem.
    **kwargs : dict
        New dataitem parameters.

    Returns
    -------
    Dataitem
        Object instance.
    """
    obj = log_dataitem(
        project=self.name,
        name=name,
        kind=kind,
        path=path,
        data=data,
        extension=extension,
        **kwargs,
    )
    self.refresh()
    return obj

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

Create a new object instance.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
uuid	str	ID of the object (UUID4).	None
description	str	Description of the object (human readable).	None
git_source	str	Remote git source for object.	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object must be embedded in project.	True
path	str	Object path on local file system or remote storage. If not provided, it's generated.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Dataitem	Object instance.

Source code in digitalhub_data/entities/projects/entity.py

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

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4).
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object must be embedded in project.
    path : str
        Object path on local file system or remote storage.
        If not provided, it's generated.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Dataitem
        Object instance.
    """
    obj = new_dataitem(
        project=self.name,
        name=name,
        kind=kind,
        path=path,
        uuid=uuid,
        description=description,
        git_source=git_source,
        labels=labels,
        embedded=embedded,
        **kwargs,
    )
    self.refresh()
    return obj

project_from_dict(obj)

Create project from dictionary.

Parameters:

Name	Type	Description	Default
obj	dict	Dictionary to create object from.	required

Returns:

Type	Description
ProjectData	ProjectData object.

Source code in digitalhub_data/entities/projects/entity.py

def project_from_dict(obj: dict) -> ProjectData:
    """
    Create project from dictionary.

    Parameters
    ----------
    obj : dict
        Dictionary to create object from.

    Returns
    -------
    ProjectData
        ProjectData object.
    """
    return ProjectData.from_dict(obj)

project_from_parameters(name, kind, description=None, git_source=None, labels=None, local=False, context=None, **kwargs)

Create project.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
description	str	Description of the object (human readable).	None
git_source	str	Remote git source for object.	None
labels	list[str]	List of labels.	None
local	bool	Flag to determine if object will be exported to backend.	False
context	str	The context of the project.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
ProjectData	ProjectData object.

Source code in digitalhub_data/entities/projects/entity.py

def project_from_parameters(
    name: str,
    kind: str,
    description: str | None = None,
    git_source: str | None = None,
    labels: list[str] | None = None,
    local: bool = False,
    context: str | None = None,
    **kwargs,
) -> ProjectData:
    """
    Create project.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    local : bool
        Flag to determine if object will be exported to backend.
    context : str
        The context of the project.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    ProjectData
        ProjectData object.
    """
    name = build_name(name)
    spec = build_spec(
        kind,
        context=context,
        **kwargs,
    )
    metadata = build_metadata(
        kind,
        project=name,
        name=name,
        description=description,
        labels=labels,
        source=git_source,
    )
    status = build_status(kind)
    return ProjectData(
        name=name,
        kind=kind,
        metadata=metadata,
        spec=spec,
        status=status,
        local=local,
    )

ProjectMl

Bases: ProjectData

ProjectMl class.

Source code in digitalhub_ml/entities/projects/entity.py

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

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

    def new_model(
        self,
        name: str,
        kind: str,
        uuid: str | None = None,
        description: str | None = None,
        git_source: str | None = None,
        labels: list[str] | None = None,
        embedded: bool = True,
        path: str | None = None,
        framework: str | None = None,
        algorithm: str | None = None,
        **kwargs,
    ) -> Model:
        """
        Create a new Model instance with the specified parameters.

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        uuid : str
            ID of the object (UUID4).
        description : str
            Description of the object (human readable).
        git_source : str
            Remote git source for object.
        labels : list[str]
            List of labels.
        embedded : bool
            Flag to determine if object must be embedded in project.
        path : str
            Object path on local file system or remote storage.
            If not provided, it's generated.
        framework : str
            Model framework (e.g. 'pytorch').
        algorithm : str
            Model algorithm (e.g. 'resnet').
        **kwargs : dict
            Spec keyword arguments.

        Returns
        -------
        Model
            An instance of the created model.
        """
        obj = new_model(
            project=self.name,
            name=name,
            kind=kind,
            uuid=uuid,
            description=description,
            git_source=git_source,
            labels=labels,
            embedded=embedded,
            path=path,
            framework=framework,
            algorithm=algorithm,
            **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 or name.
        entity_id : str
            Entity ID.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        Model
            Instance of Model class.
        """
        obj = get_model(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            **kwargs,
        )
        self.refresh()
        return obj

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

        Parameters
        ----------
        identifier : str
            Entity key or name.
        entity_id : str
            Entity ID.
        delete_all_versions : bool
            Delete all versions of the named entity.
            Use entity name instead of entity key as identifier.
        **kwargs : dict
            Parameters to pass to the API call.

        Returns
        -------
        None
        """
        delete_model(
            identifier=identifier,
            project=self.name,
            entity_id=entity_id,
            delete_all_versions=delete_all_versions,
            **kwargs,
        )
        self.refresh()

    def list_models(self, **kwargs) -> list[dict]:
        """
        List models associated with the project.

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

        Returns
        -------
        list[dict]
            List of objects related to project.
        """
        return list_models(self.name, **kwargs)

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

        Parameters
        ----------
        name : str
            Object name.
        kind : str
            Kind the object.
        path : str
            Destination path of the model.
        source_path : str
            Model location on local machine.
        **kwargs : dict
            New model parameters.

        Returns
        -------
        Model
            Instance of Model class.
        """
        obj = log_model(
            project=self.name,
            name=name,
            kind=kind,
            path=path,
            source_path=source_path,
            **kwargs,
        )
        self.refresh()
        return obj

    @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.
        """
        # Override methods to search in digitalhub_ml
        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,
        }

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

Delete a Model from project.

Parameters:

Name	Type	Description	Default
identifier	str	Entity key or name.	required
entity_id	str	Entity ID.	None
delete_all_versions	bool	Delete all versions of the named entity. Use entity name instead of entity key as identifier.	False
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
None

Source code in digitalhub_ml/entities/projects/entity.py

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

    Parameters
    ----------
    identifier : str
        Entity key or name.
    entity_id : str
        Entity ID.
    delete_all_versions : bool
        Delete all versions of the named entity.
        Use entity name instead of entity key as identifier.
    **kwargs : dict
        Parameters to pass to the API call.

    Returns
    -------
    None
    """
    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 or name.	required
entity_id	str	Entity ID.	None
**kwargs	dict	Parameters to pass to the API call.	{}

Returns:

Type	Description
Model	Instance of Model class.

Source code in digitalhub_ml/entities/projects/entity.py

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

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

    Returns
    -------
    Model
        Instance of Model class.
    """
    obj = get_model(
        identifier=identifier,
        project=self.name,
        entity_id=entity_id,
        **kwargs,
    )
    self.refresh()
    return obj

list_models(**kwargs)

List models associated with the project.

Parameters:

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

Returns:

Type	Description
list[dict]	List of objects related to project.

Source code in digitalhub_ml/entities/projects/entity.py

def list_models(self, **kwargs) -> list[dict]:
    """
    List models associated with the project.

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

    Returns
    -------
    list[dict]
        List of objects related to project.
    """
    return list_models(self.name, **kwargs)

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

Create and upload an model.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
path	str	Destination path of the model.	None
source_path	str	Model location on local machine.	None
**kwargs	dict	New model parameters.	{}

Returns:

Type	Description
Model	Instance of Model class.

Source code in digitalhub_ml/entities/projects/entity.py

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

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    path : str
        Destination path of the model.
    source_path : str
        Model location on local machine.
    **kwargs : dict
        New model parameters.

    Returns
    -------
    Model
        Instance of Model class.
    """
    obj = log_model(
        project=self.name,
        name=name,
        kind=kind,
        path=path,
        source_path=source_path,
        **kwargs,
    )
    self.refresh()
    return obj

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

Create a new Model instance with the specified parameters.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
uuid	str	ID of the object (UUID4).	None
description	str	Description of the object (human readable).	None
git_source	str	Remote git source for object.	None
labels	list[str]	List of labels.	None
embedded	bool	Flag to determine if object must be embedded in project.	True
path	str	Object path on local file system or remote storage. If not provided, it's generated.	None
framework	str	Model framework (e.g. 'pytorch').	None
algorithm	str	Model algorithm (e.g. 'resnet').	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
Model	An instance of the created model.

Source code in digitalhub_ml/entities/projects/entity.py

def new_model(
    self,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    git_source: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    path: str | None = None,
    framework: str | None = None,
    algorithm: str | None = None,
    **kwargs,
) -> Model:
    """
    Create a new Model instance with the specified parameters.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4).
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    embedded : bool
        Flag to determine if object must be embedded in project.
    path : str
        Object path on local file system or remote storage.
        If not provided, it's generated.
    framework : str
        Model framework (e.g. 'pytorch').
    algorithm : str
        Model algorithm (e.g. 'resnet').
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Model
        An instance of the created model.
    """
    obj = new_model(
        project=self.name,
        name=name,
        kind=kind,
        uuid=uuid,
        description=description,
        git_source=git_source,
        labels=labels,
        embedded=embedded,
        path=path,
        framework=framework,
        algorithm=algorithm,
        **kwargs,
    )
    self.refresh()
    return obj

project_from_dict(obj)

Create project from dictionary.

Parameters:

Name	Type	Description	Default
obj	dict	Dictionary to create object from.	required

Returns:

Type	Description
ProjectData	ProjectData object.

Source code in digitalhub_ml/entities/projects/entity.py

def project_from_dict(obj: dict) -> ProjectData:
    """
    Create project from dictionary.

    Parameters
    ----------
    obj : dict
        Dictionary to create object from.

    Returns
    -------
    ProjectData
        ProjectData object.
    """
    return ProjectMl.from_dict(obj)

project_from_parameters(name, kind, description=None, git_source=None, labels=None, local=False, context=None, **kwargs)

Create project.

Parameters:

Name	Type	Description	Default
name	str	Object name.	required
kind	str	Kind the object.	required
description	str	Description of the object (human readable).	None
git_source	str	Remote git source for object.	None
labels	list[str]	List of labels.	None
local	bool	Flag to determine if object will be exported to backend.	False
context	str	The context of the project.	None
**kwargs	dict	Spec keyword arguments.	{}

Returns:

Type	Description
ProjectData	ProjectData object.

Source code in digitalhub_ml/entities/projects/entity.py

def project_from_parameters(
    name: str,
    kind: str,
    description: str | None = None,
    git_source: str | None = None,
    labels: list[str] | None = None,
    local: bool = False,
    context: str | None = None,
    **kwargs,
) -> ProjectData:
    """
    Create project.

    Parameters
    ----------
    name : str
        Object name.
    kind : str
        Kind the object.
    description : str
        Description of the object (human readable).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    local : bool
        Flag to determine if object will be exported to backend.
    context : str
        The context of the project.
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    ProjectData
        ProjectData object.
    """
    name = build_name(name)
    spec = build_spec(
        kind,
        context=context,
        **kwargs,
    )
    metadata = build_metadata(
        kind,
        project=name,
        name=name,
        description=description,
        labels=labels,
        source=git_source,
    )
    status = build_status(kind)
    return ProjectMl(
        name=name,
        kind=kind,
        metadata=metadata,
        spec=spec,
        status=status,
        local=local,
    )