wei.types.step_types

Types related to workflow steps

Classes

StepFileResponse(status, files[, data])	Convenience wrapper for FastAPI's FileResponse class If not using FastAPI, return a response with: - The file object as the response content - The StepResponse parameters as custom headers, prefixed with "x-wei-"
StepStatus(value[, names, module, qualname, ...])	Status for a step of a workflow

pydantic model wei.types.step_types.ActionRequest

Request to perform an action on a module

Show JSON schema

{
   "title": "ActionRequest",
   "description": "Request to perform an action on a module",
   "type": "object",
   "properties": {
      "action_handle": {
         "title": "Action Handle",
         "type": "string"
      },
      "action_vars": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "title": "Action Vars"
      },
      "files": {
         "default": [],
         "items": {
            "format": "binary",
            "type": "string"
         },
         "title": "Files",
         "type": "array"
      }
   },
   "required": [
      "action_handle",
      "action_vars"
   ]
}

Config:

• use_enum_values: bool = True

Fields:

• args (Dict[str, Any] | None)

• files (List[fastapi.datastructures.UploadFile])

• name (str)

Validators:

• validate_args » args

field args: Dict[str, Any] | None [Required] (alias 'AliasChoices(choices=['action_vars', 'args'])')

Arguments for the action

Validated by:

• validate_args

• validate_ulids

field files: List[UploadFile] = []

Files to be sent along with the action

Validated by:

• validate_ulids

field name: str [Required] (alias 'AliasChoices(choices=['action_handle', 'name'])')

Name of the action to perform

Validated by:

• validate_ulids

validator validate_args  »  args

Validate the args field of the action request

pydantic model wei.types.step_types.Step

Container for a single step

Show JSON schema

{
   "title": "Step",
   "description": "Container for a single step",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "module": {
         "title": "Module",
         "type": "string"
      },
      "action": {
         "title": "Action",
         "type": "string"
      },
      "args": {
         "default": {},
         "title": "Args",
         "type": "object"
      },
      "files": {
         "additionalProperties": {
            "anyOf": [
               {
                  "type": "string"
               },
               {
                  "format": "path",
                  "type": "string"
               }
            ]
         },
         "default": {},
         "title": "Files",
         "type": "object"
      },
      "checks": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Checks"
      },
      "locations": {
         "default": {},
         "title": "Locations",
         "type": "object"
      },
      "requirements": {
         "default": {},
         "title": "Requirements",
         "type": "object"
      },
      "dependencies": {
         "default": [],
         "items": {
            "type": "string"
         },
         "title": "Dependencies",
         "type": "array"
      },
      "priority": {
         "anyOf": [
            {
               "type": "integer"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Priority"
      },
      "comment": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Comment"
      },
      "data_labels": {
         "anyOf": [
            {
               "additionalProperties": {
                  "type": "string"
               },
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Data Labels"
      },
      "id": {
         "title": "Id",
         "type": "string"
      },
      "start_time": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Start Time"
      },
      "end_time": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "End Time"
      },
      "duration": {
         "anyOf": [
            {
               "format": "duration",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Duration"
      },
      "result": {
         "anyOf": [
            {
               "$ref": "#/$defs/StepResponse"
            },
            {
               "type": "null"
            }
         ],
         "default": null
      }
   },
   "$defs": {
      "StepResponse": {
         "description": "Standard Response returned by module interfaces\nin response to action requests",
         "properties": {
            "status": {
               "$ref": "#/$defs/StepStatus",
               "default": "succeeded"
            },
            "error": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Error"
            },
            "data": {
               "anyOf": [
                  {
                     "type": "object"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Data"
            },
            "files": {
               "anyOf": [
                  {
                     "type": "object"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Files"
            }
         },
         "title": "StepResponse",
         "type": "object"
      },
      "StepStatus": {
         "description": "Status for a step of a workflow",
         "enum": [
            "idle",
            "not_ready",
            "running",
            "succeeded",
            "failed",
            "cancelled"
         ],
         "title": "StepStatus",
         "type": "string"
      }
   },
   "required": [
      "name",
      "module",
      "action"
   ]
}

Config:

• use_enum_values: bool = True

• arbitrary_types_allowed: bool = True

Fields:

• action (str)

• args (Dict[str, Any])

• checks (str | None)

• comment (str | None)

• data_labels (Dict[str, str] | None)

• dependencies (List[str])

• duration (datetime.timedelta | None)

• end_time (datetime.datetime | None)

• files (Dict[str, str | pathlib.Path])

• id (str)

• locations (Dict[str, Any])

• module (str)

• name (str)

• priority (int | None)

• requirements (Dict[str, Any])

• result (StepResponse | None)

• start_time (datetime.datetime | None)

Validators:

• validate_args_dict » args

field action: str [Required]

The command type to get executed by the robot

Validated by:

• validate_ulids

field args: Dict[str, Any] = {}

Arguments for instruction

Validated by:

• validate_args_dict

• validate_ulids

field checks: str | None = None

For future use

Validated by:

• validate_ulids

field comment: str | None = None

Notes about step

Validated by:

• validate_ulids

field data_labels: Dict[str, str] | None = None

Dictionary of user provided data labels

Validated by:

• validate_ulids

field dependencies: List[str] = []

Other steps required to be done before this can start

Validated by:

• validate_ulids

field duration: timedelta | None = None

Duration of the step’s run

Validated by:

• validate_ulids

field end_time: datetime | None = None

Time the step finished running

Validated by:

• validate_ulids

field files: Dict[str, str | Path] = {}

Files to be used in the step

Validated by:

• validate_ulids

field id: str [Optional]

ID of step

Validated by:

• validate_ulids

field locations: Dict[str, Any] = {}

locations referenced in the step

Validated by:

• validate_ulids

field module: str [Required]

Module used in the step

Validated by:

• validate_ulids

field name: str [Required]

Name of step

Validated by:

• validate_ulids

field priority: int | None = None

For scheduling

Validated by:

• validate_ulids

field requirements: Dict[str, Any] = {}

Equipment/resources needed in module

Validated by:

• validate_ulids

field result: StepResponse | None = None

Result of the step after being run

Validated by:

• validate_ulids

field start_time: datetime | None = None

Time the step started running

Validated by:

• validate_ulids

validator validate_args_dict  »  args

asserts that args dict is assembled correctly

pydantic model wei.types.step_types.StepFailed

A StepResponse for a failed step

Show JSON schema

{
   "title": "StepFailed",
   "description": "A StepResponse for a failed step",
   "type": "object",
   "properties": {
      "status": {
         "const": "failed",
         "default": "failed",
         "enum": [
            "failed"
         ],
         "title": "Status",
         "type": "string"
      },
      "error": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Error"
      },
      "data": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Data"
      },
      "files": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Files"
      }
   }
}

Config:

• use_enum_values: bool = True

Fields:

• status (Literal[wei.types.step_types.StepStatus.FAILED])

Validators:

field status: Literal[StepStatus.FAILED] = StepStatus.FAILED

Whether the step succeeded or failed

Validated by:

• validate_ulids

class wei.types.step_types.StepFileResponse(status: StepStatus, files: Dict[str, str], data: Dict[str, str] = None)

Convenience wrapper for FastAPI’s FileResponse class If not using FastAPI, return a response with: - The file object as the response content - The StepResponse parameters as custom headers, prefixed with “x-wei-”

__init__(status: StepStatus, files: Dict[str, str], data: Dict[str, str] = None)

Returns a FileResponse with the given files as the response content

pydantic model wei.types.step_types.StepResponse

Standard Response returned by module interfaces in response to action requests

Show JSON schema

{
   "title": "StepResponse",
   "description": "Standard Response returned by module interfaces\nin response to action requests",
   "type": "object",
   "properties": {
      "status": {
         "$ref": "#/$defs/StepStatus",
         "default": "succeeded"
      },
      "error": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Error"
      },
      "data": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Data"
      },
      "files": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Files"
      }
   },
   "$defs": {
      "StepStatus": {
         "description": "Status for a step of a workflow",
         "enum": [
            "idle",
            "not_ready",
            "running",
            "succeeded",
            "failed",
            "cancelled"
         ],
         "title": "StepStatus",
         "type": "string"
      }
   }
}

Config:

• use_enum_values: bool = True

Fields:

• data (Dict[str, Any] | None)

• error (str | None)

• files (Dict[str, Any] | None)

• status (wei.types.step_types.StepStatus)

Validators:

field data: Dict[str, Any] | None = None

Key value dict of data returned from step

Validated by:

• validate_ulids

field error: str | None = None

Error message resulting from the action

Validated by:

• validate_ulids

field files: Dict[str, Any] | None = None

Key value dict of file labels and file names from step

Validated by:

• validate_ulids

field status: StepStatus = StepStatus.SUCCEEDED

Whether the step succeeded or failed

Validated by:

• validate_ulids

classmethod from_headers(headers: Dict[str, Any]) → StepResponse

Creates a StepResponse from the headers of a file response

classmethod step_failed(error: str = '') → StepResponse

Returns a StepResponse for a failed step

classmethod step_not_ready(error: str = '') → StepResponse

Returns a StepResponse for a failed step

classmethod step_succeeded(files: Dict[str, str] = None, data: Dict[str, str] = None) → StepResponse

Returns a StepResponse for a successful step

to_headers() → Dict[str, str]

Converts the response to a dictionary of headers

class wei.types.step_types.StepStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Status for a step of a workflow

pydantic model wei.types.step_types.StepSucceeded

A StepResponse for a successful step

Show JSON schema

{
   "title": "StepSucceeded",
   "description": "A StepResponse for a successful step",
   "type": "object",
   "properties": {
      "status": {
         "const": "succeeded",
         "default": "succeeded",
         "enum": [
            "succeeded"
         ],
         "title": "Status",
         "type": "string"
      },
      "error": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Error"
      },
      "data": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Data"
      },
      "files": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Files"
      }
   }
}

Config:

• use_enum_values: bool = True

Fields:

• status (Literal[wei.types.step_types.StepStatus.SUCCEEDED])

Validators:

field status: Literal[StepStatus.SUCCEEDED] = StepStatus.SUCCEEDED

Whether the step succeeded or failed

Validated by:

• validate_ulids