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.

PARAMETER DESCRIPTION
package_manifest

TYPE: ManifestV2

package_root

TYPE: Path

package_version

TYPE: str | None DEFAULT: None

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 
 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
142
143
144
145
146
147
148
149
150
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[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 = {}
        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 = TaskCreate(
            **_task.model_dump(
                exclude={
                    "executable_non_parallel",
                    "executable_parallel",
                }
            ),
            **task_attributes,
            authors=package_manifest.authors,
        )
        task_list.append(task_obj)
    return task_list