Skip to content

task_collection

TaskCollectPipV1

Bases: BaseModel

TaskCollectPip 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 path of a local wheel file;
2. `package` is the name of a package that can be installed via `pip`.

Attributes:

Name Type Description
package str

The name of a pip-installable package, or the path to a local wheel file.
package_version Optional[str]

Version of the package
package_extras Optional[str]

Package extras to include in the pip install command
python_version Optional[str]

Python version to install and run the package tasks
pinned_package_versions Optional[dict[str, str]]

dictionary 'package':'version' used to pin versions for specific packages.
Source code in fractal_server/app/schemas/v1/task_collection.py 
18
19
20
21
22
23
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
class TaskCollectPipV1(BaseModel):
    """
    TaskCollectPip 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 path of a local wheel file;
        2. `package` is the name of a package that can be installed via `pip`.


    Attributes:
        package:
            The name of a `pip`-installable package, or the path to a local
            wheel file.
        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:
            dictionary 'package':'version' used to pin versions for specific
            packages.

    """

    package: str
    package_version: Optional[str] = None
    package_extras: Optional[str] = None
    python_version: Optional[str] = None
    pinned_package_versions: Optional[dict[str, str]] = None

    _package_extras = validator("package_extras", allow_reuse=True)(
        valstr("package_extras")
    )
    _python_version = validator("python_version", allow_reuse=True)(
        valstr("python_version")
    )

    @validator("package")
    def package_validator(cls, value):
        if "/" in value:
            if not value.endswith(".whl"):
                raise ValueError(
                    "Local-package path must be a wheel file "
                    f"(given {value})."
                )
            if not Path(value).is_absolute():
                raise ValueError(
                    f"Local-package path must be absolute: (given {value})."
                )
        return value

    @validator("package_version")
    def package_version_validator(cls, v, values):

        valstr("package_version")(v)

        if values["package"].endswith(".whl"):
            raise ValueError(
                "Cannot provide version when package is a Wheel file."
            )
        return v

TaskCollectStatusV1

Bases: BaseModel

TaskCollectStatus class

Attributes:

Name Type Description
status Literal['pending', 'installing', 'collecting', 'fail', 'OK']
package str
venv_path Path
task_list Optional[list[TaskReadV1]]
log Optional[str]
info Optional[str]
Source code in fractal_server/app/schemas/v1/task_collection.py 
 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
class TaskCollectStatusV1(BaseModel):
    """
    TaskCollectStatus class

    Attributes:
        status:
        package:
        venv_path:
        task_list:
        log:
        info:
    """

    status: Literal["pending", "installing", "collecting", "fail", "OK"]
    package: str
    venv_path: Path
    task_list: Optional[list[TaskReadV1]] = Field(default=[])
    log: Optional[str]
    info: Optional[str]

    def sanitised_dict(self):
        """
        Return `self.dict()` after casting `self.venv_path` to a string
        """
        d = self.dict()
        d["venv_path"] = str(self.venv_path)
        return d

sanitised_dict()

Return self.dict() after casting self.venv_path to a string

Source code in fractal_server/app/schemas/v1/task_collection.py 
104
105
106
107
108
109
110
def sanitised_dict(self):
    """
    Return `self.dict()` after casting `self.venv_path` to a string
    """
    d = self.dict()
    d["venv_path"] = str(self.venv_path)
    return d