Skip to content

Utils background

FUNCTION DESCRIPTION
prepare_tasks_metadata

Based on the package manifest and additional info, prepare the task list.

Classes

Functions:

prepare_tasks_metadata(*, package_manifest, package_root, python_bin=None, project_python_wrapper=None, package_version)

Based on the package manifest and additional info, prepare the task list.

PARAMETER DESCRIPTION

package_manifest

TYPE: ManifestV2

package_root

TYPE: Path

package_version

TYPE: str

python_bin

TYPE: Path | None DEFAULT: None

project_python_wrapper

TYPE: Path | None DEFAULT: None

Source code in fractal_server/tasks/v2/utils_background.py
def prepare_tasks_metadata(
    *,
    package_manifest: ManifestV2,
    package_root: Path,
    python_bin: Path | None = None,
    project_python_wrapper: Path | None = None,
    package_version: str,
) -> list[TaskCreate]:
    """
    Based on the package manifest and additional info, prepare the task list.

    Args:
        package_manifest:
        package_root:
        package_version:
        python_bin:
        project_python_wrapper:
    """

    if python_bin is not None and project_python_wrapper is None:
        actual_python = python_bin
    elif python_bin is None and project_python_wrapper is not None:
        actual_python = project_python_wrapper
    else:
        raise UnreachableBranchError(
            f"Either {project_python_wrapper} or {python_bin} must be set."
        )

    task_list = []
    for _task in package_manifest.task_list:
        # Set non-command attributes
        task_attributes = {}
        task_attributes["version"] = package_version
        if package_manifest.has_args_schemas:
            task_attributes["args_schema_version"] = (
                package_manifest.args_schema_version
            )
        # Set command attributes
        if _task.executable_non_parallel is not None:
            non_parallel_path = package_root / _task.executable_non_parallel
            cmd_non_parallel = (
                f"{actual_python.as_posix()} {non_parallel_path.as_posix()}"
            )
            task_attributes["command_non_parallel"] = cmd_non_parallel
        if _task.executable_parallel is not None:
            parallel_path = package_root / _task.executable_parallel
            cmd_parallel = (
                f"{actual_python.as_posix()} {parallel_path.as_posix()}"
            )
            task_attributes["command_parallel"] = cmd_parallel
        # Create object
        task_obj = TaskCreate(
            **_task.model_dump(
                exclude={
                    "executable_non_parallel",
                    "executable_parallel",
                }
            ),
            **task_attributes,
            authors=package_manifest.authors,
        )
        task_list.append(task_obj)
    return task_list