Skip to content

utils_background

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

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

Parameters:

Name Type Description Default
package_manifest ManifestV2
required
package_root Path
required
package_version str | None
None
python_bin Path | None
None
project_python_wrapper Path | None
None
Source code in fractal_server/tasks/v2/utils_background.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def prepare_tasks_metadata(
    *,
    package_manifest: ManifestV2,
    package_root: Path,
    python_bin: Path | None = None,
    project_python_wrapper: Path | None = None,
    package_version: str | None = None,
) -> list[TaskCreateV2]:
    """
    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 bool(project_python_wrapper is None) == bool(python_bin is None):
        raise UnreachableBranchError(
            f"Either {project_python_wrapper} or {python_bin} must be set."
        )

    if python_bin is not None:
        actual_python = python_bin
    else:
        actual_python = project_python_wrapper

    task_list = []
    for _task in package_manifest.task_list:
        # Set non-command attributes
        task_attributes = {}
        if package_version is not None:
            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 = TaskCreateV2(
            **_task.model_dump(
                exclude={
                    "executable_non_parallel",
                    "executable_parallel",
                }
            ),
            **task_attributes,
            authors=package_manifest.authors,
        )
        task_list.append(task_obj)
    return task_list