Skip to content

task_collection

FractalUploadedFile

Bases: BaseModel

Model for data sent from the endpoint to the background task.

Source code in fractal_server/app/schemas/v2/task_collection.py 
15
16
17
18
19
20
21
class FractalUploadedFile(BaseModel):
    """
    Model for data sent from the endpoint to the background task.
    """

    filename: str
    contents: bytes

TaskCollectCustomV2

Bases: BaseModel

Attributes:

Name Type Description
manifest ManifestV2

Manifest of a Fractal task package (this is typically the content of __FRACTAL_MANIFEST__.json).
python_interpreter AbsolutePathStr

Absolute path to the Python interpreter to be used for running tasks.
name AbsolutePathStr

A name identifying this package, that will fill the TaskGroupV2.pkg_name column.
package_root AbsolutePathStr | None

The folder where the package is installed. If not provided, it will be extracted via pip show (requires package_name to be set).
package_name NonEmptyStr | None

Name of the package, as used for import <package_name>; this is then used to extract the package directory (package_root) via pip show <package_name>.
version NonEmptyStr | None

Optional version of tasks to be collected.
Source code in fractal_server/app/schemas/v2/task_collection.py 
 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
class TaskCollectCustomV2(BaseModel):
    """
    Attributes:
        manifest: Manifest of a Fractal task package (this is typically the
            content of `__FRACTAL_MANIFEST__.json`).
        python_interpreter: Absolute path to the Python interpreter to be used
            for running tasks.
        name: A name identifying this package, that will fill the
            `TaskGroupV2.pkg_name` column.
        package_root: The folder where the package is installed.
            If not provided, it will be extracted via `pip show`
            (requires `package_name` to be set).
        package_name: Name of the package, as used for `import <package_name>`;
            this is then used to extract the package directory (`package_root`)
            via `pip show <package_name>`.
        version: Optional version of tasks to be collected.
    """

    model_config = ConfigDict(extra="forbid")
    manifest: ManifestV2
    python_interpreter: AbsolutePathStr
    label: NonEmptyStr
    package_root: AbsolutePathStr | None = None
    package_name: NonEmptyStr | None = None
    version: NonEmptyStr | None = None

    @field_validator("package_name", mode="after")
    @classmethod
    def validate_package_name(cls, value):
        if value is not None:
            validate_cmd(value)
        return value

    @model_validator(mode="before")
    @classmethod
    def one_of_package_root_or_name(cls, values):
        package_root = values.get("package_root")
        package_name = values.get("package_name")
        if (package_root is None and package_name is None) or (
            package_root is not None and package_name is not None
        ):
            raise ValueError(
                "One and only one must be set between "
                "'package_root' and 'package_name'"
            )
        return values

TaskCollectPipV2

Bases: BaseModel

TaskCollectPipV2 class

This class only encodes the attributes required to trigger a task-collection operation. Other attributes (that are assigned during task collection) are defined as part of fractal-server.

Two cases are supported:

1. `package` is the name of a package that can be installed via `pip`.
1. `package=None`, and a wheel file is uploaded within the API request.

Attributes:

Name Type Description
package NonEmptyStr | None

The name of a pip-installable package, or None.
package_version NonEmptyStr | None

Version of the package
package_extras NonEmptyStr | None

Package extras to include in the pip install command
python_version Literal['3.9', '3.10', '3.11', '3.12', '3.13'] | None

Python version to install and run the package tasks
pinned_package_versions_pre DictStrStr | None

dictionary 'package':'version' used to pre-pin versions for specific packages.
pinned_package_versions_post DictStrStr | None

dictionary 'package':'version' used to post-pin versions for specific packages.
Source code in fractal_server/app/schemas/v2/task_collection.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class TaskCollectPipV2(BaseModel):
    """
    TaskCollectPipV2 class

    This class only encodes the attributes required to trigger a
    task-collection operation. Other attributes (that are assigned *during*
    task collection) are defined as part of fractal-server.

    Two cases are supported:

        1. `package` is the name of a package that can be installed via `pip`.
        1. `package=None`, and a wheel file is uploaded within the API request.

    Attributes:
        package: The name of a `pip`-installable package, or `None`.
        package_version: Version of the package
        package_extras: Package extras to include in the `pip install` command
        python_version: Python version to install and run the package tasks
        pinned_package_versions_pre:
            dictionary 'package':'version' used to pre-pin versions for
            specific packages.
        pinned_package_versions_post:
            dictionary 'package':'version' used to post-pin versions for
            specific packages.
    """

    model_config = ConfigDict(extra="forbid")
    package: NonEmptyStr | None = None
    package_version: NonEmptyStr | None = None
    package_extras: NonEmptyStr | None = None
    python_version: Literal[
        "3.9", "3.10", "3.11", "3.12", "3.13"
    ] | None = None
    pinned_package_versions_pre: DictStrStr | None = None
    pinned_package_versions_post: DictStrStr | None = None

    @field_validator(
        "package",
        "package_version",
        "package_extras",
        mode="after",
    )
    @classmethod
    def validate_commands(cls, value):
        if value is not None:
            validate_cmd(value)
        return value

    @field_validator(
        "pinned_package_versions_pre",
        "pinned_package_versions_post",
        mode="after",
    )
    @classmethod
    def validate_pinned_package_versions(cls, value):
        if value is not None:
            for pkg, version in value.items():
                validate_cmd(pkg)
                validate_cmd(version)
        return value