Skip to content

workflowtask

WorkflowTaskImportV2

Bases: BaseModel

Source code in fractal_server/app/schemas/v2/workflowtask.py 
 69
 70
 71
 72
 73
 74
 75
 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
class WorkflowTaskImportV2(BaseModel):
    model_config = ConfigDict(extra="forbid")

    meta_non_parallel: Optional[DictStrAny] = None
    meta_parallel: Optional[DictStrAny] = None
    args_non_parallel: Optional[DictStrAny] = None
    args_parallel: Optional[DictStrAny] = None
    type_filters: Optional[TypeFilters] = None
    input_filters: Optional[dict[str, Any]] = None

    task: Union[TaskImportV2, TaskImportV2Legacy]

    @model_validator(mode="before")
    @classmethod
    def update_legacy_filters(cls, values: 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

update_legacy_filters(values) classmethod

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

Source code in fractal_server/app/schemas/v2/workflowtask.py 
 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
@model_validator(mode="before")
@classmethod
def update_legacy_filters(cls, values: 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

WorkflowTaskReplaceV2

Bases: BaseModel

Used by 'replace-task' endpoint

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

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