Skip to content

Workflowtask

CLASS DESCRIPTION
WorkflowTaskImport
WorkflowTaskReplace

Used by 'replace-task' endpoint

Attributes

Classes

WorkflowTaskImport pydantic-model

Bases: BaseModel

Config:

  • extra: forbid

Fields:

Validators:

Source code in fractal_server/app/schemas/v2/workflowtask.py 
 76
 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
class WorkflowTaskImport(BaseModel):
    model_config = ConfigDict(extra="forbid")

    meta_non_parallel: DictStrAny | None = None
    meta_parallel: DictStrAny | None = None
    args_non_parallel: DictStrAny | None = None
    args_parallel: DictStrAny | None = None
    type_filters: TypeFilters | None = None
    input_filters: dict[str, Any] | None = None
    description: NonEmptyStr | None = None
    alias: NonEmptyStr | None = None

    task: TaskImport

    @model_validator(mode="before")
    @classmethod
    def update_legacy_filters(cls, values: dict) -> dict:
        """
        Transform legacy filters (created with fractal-server<2.11.0)
        into type filters
        """
        if values.get("input_filters") is not None:
            if "type_filters" in values.keys():
                raise ValueError(
                    "Cannot set filters both through the legacy field "
                    "('filters') and the new one ('type_filters')."
                )
            else:
                # As of 2.11.0, WorkflowTask do not have attribute filters
                # any more.
                if values["input_filters"]["attributes"] != {}:
                    raise ValueError(
                        "Cannot set attribute filters for WorkflowTasks."
                    )
                # Convert legacy filters.types into new type_filters
                values["type_filters"] = values["input_filters"].get(
                    "types", {}
                )
                values["input_filters"] = None

        return values

Methods:

update_legacy_filters(values) pydantic-validator

Transform legacy filters (created with fractal-server<2.11.0) into type filters

Source code in fractal_server/app/schemas/v2/workflowtask.py 
 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
@model_validator(mode="before")
@classmethod
def update_legacy_filters(cls, values: dict) -> dict:
    """
    Transform legacy filters (created with fractal-server<2.11.0)
    into type filters
    """
    if values.get("input_filters") is not None:
        if "type_filters" in values.keys():
            raise ValueError(
                "Cannot set filters both through the legacy field "
                "('filters') and the new one ('type_filters')."
            )
        else:
            # As of 2.11.0, WorkflowTask do not have attribute filters
            # any more.
            if values["input_filters"]["attributes"] != {}:
                raise ValueError(
                    "Cannot set attribute filters for WorkflowTasks."
                )
            # Convert legacy filters.types into new type_filters
            values["type_filters"] = values["input_filters"].get(
                "types", {}
            )
            values["input_filters"] = None

    return values

WorkflowTaskReplace pydantic-model

Bases: BaseModel

Used by 'replace-task' endpoint

Fields:

Source code in fractal_server/app/schemas/v2/workflowtask.py 
32
33
34
35
36
class WorkflowTaskReplace(BaseModel):
    """Used by 'replace-task' endpoint"""

    args_non_parallel: dict[str, Any] | None = None
    args_parallel: dict[str, Any] | None = None