Entity and methods

Function

Bases: Entity

A class representing a function.

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

class Function(Entity):
    """
    A class representing a function.
    """

    ENTITY_TYPE = EntityTypes.FUNCTIONS.value

    def __init__(
        self,
        project: str,
        name: str,
        uuid: str,
        kind: str,
        metadata: Metadata,
        spec: FunctionSpec,
        status: FunctionStatus,
        user: str | None = None,
    ) -> None:
        """
        Constructor.

        Parameters
        ----------
        project : str
            Project name.
        name : str
            Name of the object.
        uuid : str
            Version of the object.
        kind : str
            Kind the object.
        metadata : Metadata
            Metadata of the object.
        spec : FunctionSpec
            Specification of the object.
        status : FunctionStatus
            Status of the object.
        user : str
            Owner of the object.
        """
        super().__init__()
        self.project = project
        self.name = name
        self.id = uuid
        self.kind = kind
        self.key = f"store://{project}/{self.ENTITY_TYPE}/{kind}/{name}:{uuid}"
        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(["project", "name", "id", "key"])

        # Initialize tasks
        self._tasks: dict[str, Task] = {}

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

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

        Parameters
        ----------
        update : bool
            Flag to indicate update.

        Returns
        -------
        Function
            Entity saved.
        """
        obj = self.to_dict(include_all_non_private=True)

        if not update:
            new_obj = create_entity_api_ctx(self.project, self.ENTITY_TYPE, obj)
            self._update_attributes(new_obj)
            return self

        self.metadata.updated = obj["metadata"]["updated"] = get_timestamp()
        new_obj = update_entity_api_ctx(self.project, self.ENTITY_TYPE, self.id, obj)
        self._update_attributes(new_obj)
        return self

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

        Returns
        -------
        Function
            Entity refreshed.
        """
        new_obj = read_entity_api_ctx(self.key)
        self._update_attributes(new_obj)
        return self

    def export(self, filename: str | None = None) -> None:
        """
        Export object 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.to_dict()

        if filename is None:
            filename = f"function_{self.kind}_{self.name}_{self.id}.yml"
        pth = self._context().root / filename
        pth.parent.mkdir(parents=True, exist_ok=True)

        # Embed tasks in file
        if self._tasks:
            obj = [obj] + [v.to_dict() for _, v in self._tasks.items()]

        write_yaml(pth, obj)

    #############################
    #  Context
    #############################

    def _context(self) -> Context:
        """
        Get context.

        Returns
        -------
        Context
            Context.
        """
        return get_context(self.project)

    #############################
    #  Function Methods
    #############################

    def run(
        self,
        action: str,
        local_execution: bool = False,
        **kwargs,
    ) -> Run:
        """
        Run function. This method creates a new run and executes it.

        Parameters
        ----------
        action : str
            Action to execute.
        local_execution : bool
            Flag to determine if object has local execution.
        **kwargs : dict
            Keyword arguments passed to Task and Run builders.

        Returns
        -------
        Run
            Run instance.
        """
        # Get kind registry
        kind_reg = get_kind_registry(self.kind)

        # Get task and run kind
        task_kind = kind_reg.get_task_kind_from_action(action=action)
        run_kind = kind_reg.get_run_kind()

        # Create or update new task
        task = self.new_task(task_kind, **kwargs)

        # Run function from task
        run = task.run(run_kind, local_execution, **kwargs)

        # If execution is done by DHCore backend, return the object
        if not local_execution:
            if self._context().local:
                raise BackendError("Cannot run remote function with local backend.")
            return run

        # If local execution, build and launch run.
        # Detach the run from the main thread
        run.build()
        with ThreadPoolExecutor(max_workers=1) as executor:
            result = executor.submit(run.run)
            r = result.result()
        return r

    #############################
    #  Helpers
    #############################

    def _get_function_string(self) -> str:
        """
        Get function string.

        Returns
        -------
        str
            Function string.
        """
        return f"{self.kind}://{self.project}/{self.name}:{self.id}"

    def import_tasks(self, tasks: list[dict]) -> None:
        """
        Import tasks from yaml.

        Parameters
        ----------
        tasks : list[dict]
            List of tasks to import.

        Returns
        -------
        None
        """
        # Loop over tasks list, in the case where the function
        # is imported from local file.
        for task in tasks:
            # If task is not a dictionary, skip it
            if not isinstance(task, dict):
                continue

            # Create the object instance from dictionary,
            # the form in which tasks are stored in function
            # status
            task_obj = create_task_from_dict(task)

            # Try to save it in backend to been able to use
            # it for launching runs. In fact, tasks must be
            # persisted in backend to be able to launch runs.
            # Ignore if task already exists
            try:
                task_obj.save()
            except BackendError:
                pass

            # Set task if function is the same. Overwrite
            # status task dict with the new task object
            if task_obj.spec.function == self._get_function_string():
                self._tasks[task_obj.kind] = task_obj

    def new_task(self, task_kind: str, **kwargs) -> Task:
        """
        Create new task.

        Parameters
        ----------
        task_kind : str
            Kind the object.
        **kwargs : dict
            Keyword arguments.

        Returns
        -------
        Task
            New task.
        """
        # Override kwargs
        kwargs["project"] = self.project
        kwargs["function"] = self._get_function_string()
        kwargs["kind"] = task_kind

        # Create object instance
        task = create_task(**kwargs)

        exists, task_id = self._check_task_in_backend(task_kind)

        # Save or update task
        if not exists:
            task.save()
        else:
            task.id = task_id
            task.save(update=True)

        self._tasks[task_kind] = task
        return task

    def update_task(self, kind: str, **kwargs) -> None:
        """
        Update task.

        Parameters
        ----------
        kind : str
            Kind the object.
        **kwargs : dict
            Keyword arguments.

        Returns
        -------
        None

        Raises
        ------
        EntityError
            If task does not exist.
        """
        self._raise_if_not_exists(kind)

        # Update kwargs
        kwargs["project"] = self.project
        kwargs["kind"] = kind
        kwargs["function"] = self._get_function_string()
        kwargs["uuid"] = self._tasks[kind].id

        # Update task
        task = create_task(**kwargs)
        task.save(update=True)
        self._tasks[kind] = task

    def get_task(self, kind: str) -> Task:
        """
        Get task.

        Parameters
        ----------
        kind : str
            Kind the object.

        Returns
        -------
        Task
            Task.

        Raises
        ------
        EntityError
            If task is not created.
        """
        self._raise_if_not_exists(kind)
        return self._tasks[kind]

    def delete_task(self, kind: str, cascade: bool = True) -> None:
        """
        Delete task.

        Parameters
        ----------
        kind : str
            Kind the object.
        cascade : bool
            Flag to determine if cascade deletion must be performed.

        Returns
        -------
        None

        Raises
        ------
        EntityError
            If task is not created.
        """
        self._raise_if_not_exists(kind)
        delete_task(self._tasks[kind].key, cascade=cascade)
        self._tasks.pop(kind, None)

    def _check_task_in_backend(self, kind: str) -> tuple[bool, str | None]:
        """
        Check if task exists in backend.

        Parameters
        ----------
        kind : str
            Kind the object.

        Returns
        -------
        tuple[bool, str | None]
            Flag to determine if task exists in backend and ID if exists.
        """
        # List tasks from backend filtered by function and kind
        params = {"function": self._get_function_string(), "kind": kind}
        objs = list_entity_api_ctx(self.project, EntityTypes.TASKS.value, params=params)
        try:
            return True, objs[0]["id"]
        except IndexError:
            return False, None

    def _raise_if_not_exists(self, kind: str) -> None:
        """
        Raise error if task is not created.

        Parameters
        ----------
        kind : str
            Kind the object.

        Returns
        -------
        None

        Raises
        ------
        EntityError
            If task does not exist.
        """
        if self._tasks.get(kind) is None:
            raise EntityError("Task does not exist.")

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

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

        Parameters
        ----------
        obj : dict
            Dictionary to parse.
        validate : bool
            Flag to determine if validation must be performed.

        Returns
        -------
        dict
            A dictionary containing the attributes of the entity instance.
        """
        project = obj.get("project")
        name = build_name(obj.get("name"))
        kind = obj.get("kind")
        uuid = build_uuid(obj.get("id"))
        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")
        return {
            "project": project,
            "name": name,
            "uuid": uuid,
            "kind": kind,
            "metadata": metadata,
            "spec": spec,
            "status": status,
            "user": user,
        }

__init__(project, name, uuid, kind, metadata, spec, status, user=None)

Constructor.

Parameters:

Name	Type	Description	Default
project	str	Project name.	required
name	str	Name of the object.	required
uuid	str	Version of the object.	required
kind	str	Kind the object.	required
metadata	Metadata	Metadata of the object.	required
spec	FunctionSpec	Specification of the object.	required
status	FunctionStatus	Status of the object.	required
user	str	Owner of the object.	None

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

def __init__(
    self,
    project: str,
    name: str,
    uuid: str,
    kind: str,
    metadata: Metadata,
    spec: FunctionSpec,
    status: FunctionStatus,
    user: str | None = None,
) -> None:
    """
    Constructor.

    Parameters
    ----------
    project : str
        Project name.
    name : str
        Name of the object.
    uuid : str
        Version of the object.
    kind : str
        Kind the object.
    metadata : Metadata
        Metadata of the object.
    spec : FunctionSpec
        Specification of the object.
    status : FunctionStatus
        Status of the object.
    user : str
        Owner of the object.
    """
    super().__init__()
    self.project = project
    self.name = name
    self.id = uuid
    self.kind = kind
    self.key = f"store://{project}/{self.ENTITY_TYPE}/{kind}/{name}:{uuid}"
    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(["project", "name", "id", "key"])

    # Initialize tasks
    self._tasks: dict[str, Task] = {}

delete_task(kind, cascade=True)

Delete task.

Parameters:

Name	Type	Description	Default
kind	str	Kind the object.	required
cascade	bool	Flag to determine if cascade deletion must be performed.	True

Returns:

Type	Description
None

Raises:

Type	Description
EntityError	If task is not created.

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

def delete_task(self, kind: str, cascade: bool = True) -> None:
    """
    Delete task.

    Parameters
    ----------
    kind : str
        Kind the object.
    cascade : bool
        Flag to determine if cascade deletion must be performed.

    Returns
    -------
    None

    Raises
    ------
    EntityError
        If task is not created.
    """
    self._raise_if_not_exists(kind)
    delete_task(self._tasks[kind].key, cascade=cascade)
    self._tasks.pop(kind, None)

export(filename=None)

Export object 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/functions/entity.py

def export(self, filename: str | None = None) -> None:
    """
    Export object 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.to_dict()

    if filename is None:
        filename = f"function_{self.kind}_{self.name}_{self.id}.yml"
    pth = self._context().root / filename
    pth.parent.mkdir(parents=True, exist_ok=True)

    # Embed tasks in file
    if self._tasks:
        obj = [obj] + [v.to_dict() for _, v in self._tasks.items()]

    write_yaml(pth, obj)

get_task(kind)

Get task.

Parameters:

Name	Type	Description	Default
kind	str	Kind the object.	required

Returns:

Type	Description
Task	Task.

Raises:

Type	Description
EntityError	If task is not created.

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

def get_task(self, kind: str) -> Task:
    """
    Get task.

    Parameters
    ----------
    kind : str
        Kind the object.

    Returns
    -------
    Task
        Task.

    Raises
    ------
    EntityError
        If task is not created.
    """
    self._raise_if_not_exists(kind)
    return self._tasks[kind]

import_tasks(tasks)

Import tasks from yaml.

Parameters:

Name	Type	Description	Default
tasks	list[dict]	List of tasks to import.	required

Returns:

Type	Description
None

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

def import_tasks(self, tasks: list[dict]) -> None:
    """
    Import tasks from yaml.

    Parameters
    ----------
    tasks : list[dict]
        List of tasks to import.

    Returns
    -------
    None
    """
    # Loop over tasks list, in the case where the function
    # is imported from local file.
    for task in tasks:
        # If task is not a dictionary, skip it
        if not isinstance(task, dict):
            continue

        # Create the object instance from dictionary,
        # the form in which tasks are stored in function
        # status
        task_obj = create_task_from_dict(task)

        # Try to save it in backend to been able to use
        # it for launching runs. In fact, tasks must be
        # persisted in backend to be able to launch runs.
        # Ignore if task already exists
        try:
            task_obj.save()
        except BackendError:
            pass

        # Set task if function is the same. Overwrite
        # status task dict with the new task object
        if task_obj.spec.function == self._get_function_string():
            self._tasks[task_obj.kind] = task_obj

new_task(task_kind, **kwargs)

Create new task.

Parameters:

Name	Type	Description	Default
task_kind	str	Kind the object.	required
**kwargs	dict	Keyword arguments.	{}

Returns:

Type	Description
Task	New task.

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

def new_task(self, task_kind: str, **kwargs) -> Task:
    """
    Create new task.

    Parameters
    ----------
    task_kind : str
        Kind the object.
    **kwargs : dict
        Keyword arguments.

    Returns
    -------
    Task
        New task.
    """
    # Override kwargs
    kwargs["project"] = self.project
    kwargs["function"] = self._get_function_string()
    kwargs["kind"] = task_kind

    # Create object instance
    task = create_task(**kwargs)

    exists, task_id = self._check_task_in_backend(task_kind)

    # Save or update task
    if not exists:
        task.save()
    else:
        task.id = task_id
        task.save(update=True)

    self._tasks[task_kind] = task
    return task

refresh()

Refresh object from backend.

Returns:

Type	Description
Function	Entity refreshed.

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

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

    Returns
    -------
    Function
        Entity refreshed.
    """
    new_obj = read_entity_api_ctx(self.key)
    self._update_attributes(new_obj)
    return self

run(action, local_execution=False, **kwargs)

Run function. This method creates a new run and executes it.

Parameters:

Name	Type	Description	Default
action	str	Action to execute.	required
local_execution	bool	Flag to determine if object has local execution.	False
**kwargs	dict	Keyword arguments passed to Task and Run builders.	{}

Returns:

Type	Description
Run	Run instance.

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

def run(
    self,
    action: str,
    local_execution: bool = False,
    **kwargs,
) -> Run:
    """
    Run function. This method creates a new run and executes it.

    Parameters
    ----------
    action : str
        Action to execute.
    local_execution : bool
        Flag to determine if object has local execution.
    **kwargs : dict
        Keyword arguments passed to Task and Run builders.

    Returns
    -------
    Run
        Run instance.
    """
    # Get kind registry
    kind_reg = get_kind_registry(self.kind)

    # Get task and run kind
    task_kind = kind_reg.get_task_kind_from_action(action=action)
    run_kind = kind_reg.get_run_kind()

    # Create or update new task
    task = self.new_task(task_kind, **kwargs)

    # Run function from task
    run = task.run(run_kind, local_execution, **kwargs)

    # If execution is done by DHCore backend, return the object
    if not local_execution:
        if self._context().local:
            raise BackendError("Cannot run remote function with local backend.")
        return run

    # If local execution, build and launch run.
    # Detach the run from the main thread
    run.build()
    with ThreadPoolExecutor(max_workers=1) as executor:
        result = executor.submit(run.run)
        r = result.result()
    return r

save(update=False)

Save entity into backend.

Parameters:

Name	Type	Description	Default
update	bool	Flag to indicate update.	False

Returns:

Type	Description
Function	Entity saved.

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

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

    Parameters
    ----------
    update : bool
        Flag to indicate update.

    Returns
    -------
    Function
        Entity saved.
    """
    obj = self.to_dict(include_all_non_private=True)

    if not update:
        new_obj = create_entity_api_ctx(self.project, self.ENTITY_TYPE, obj)
        self._update_attributes(new_obj)
        return self

    self.metadata.updated = obj["metadata"]["updated"] = get_timestamp()
    new_obj = update_entity_api_ctx(self.project, self.ENTITY_TYPE, self.id, obj)
    self._update_attributes(new_obj)
    return self

update_task(kind, **kwargs)

Update task.

Parameters:

Name	Type	Description	Default
kind	str	Kind the object.	required
**kwargs	dict	Keyword arguments.	{}

Returns:

Type	Description
None

Raises:

Type	Description
EntityError	If task does not exist.

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

def update_task(self, kind: str, **kwargs) -> None:
    """
    Update task.

    Parameters
    ----------
    kind : str
        Kind the object.
    **kwargs : dict
        Keyword arguments.

    Returns
    -------
    None

    Raises
    ------
    EntityError
        If task does not exist.
    """
    self._raise_if_not_exists(kind)

    # Update kwargs
    kwargs["project"] = self.project
    kwargs["kind"] = kind
    kwargs["function"] = self._get_function_string()
    kwargs["uuid"] = self._tasks[kind].id

    # Update task
    task = create_task(**kwargs)
    task.save(update=True)
    self._tasks[kind] = task

function_from_dict(obj)

Create function from dictionary.

Parameters:

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

Returns:

Type	Description
Function	Function object.

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

def function_from_dict(obj: dict) -> Function:
    """
    Create function from dictionary.

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

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

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

Create a new Function instance and persist it to the backend.

Parameters:

Name	Type	Description	Default
project	str	Project name.	required
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/functions/entity.py

def function_from_parameters(
    project: str,
    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 new Function instance and persist it to the backend.

    Parameters
    ----------
    project : str
        Project name.
    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.
    """
    name = build_name(name)
    uuid = build_uuid(uuid)
    spec = build_spec(kind, **kwargs)
    metadata = build_metadata(
        kind,
        project=project,
        name=name,
        version=uuid,
        description=description,
        source=git_source,
        labels=labels,
        embedded=embedded,
    )
    status = build_status(kind)
    return Function(
        project=project,
        name=name,
        uuid=uuid,
        kind=kind,
        metadata=metadata,
        spec=spec,
        status=status,
    )