Workunit Definition

Workunit Definition#

The idea of the workunit definition is to provide a persistable and comprehensive description of a workunit. To keep the logic even more modular it is separated into two components, the execution and the registration information.

Creating WorkunitDefinition instances#

The WorkunitDefinition class is a Pydantic model and can be created by passing a dictionary to the constructor. However, for convenience and easier integration into command line tools there is a constructor for both creating an instance from a Bfabric entity, and parsing a YAML file which contains a persisted version of the workunit

Workunit references#

Several functions and command line tools allow providing a “workunit reference”. This means, that either the ID or a path to a local YAML file can be passed to this function. If the input is a path, then the persisted information will be retrieved to instantiate a WorkunitDefinition instance, whereas if it is an integer, the information will be obtained by querying the B-Fabric API.

Since in some workflows the workunit will be used several times, and in particular not necessarily in the same process, the usual entity caching mechanism might not be able to cache the requests. Therefore, in many cases passing a reference to a YAML file is the preferred way to provide the workunit information, as it will reduce the number of requests to the B-Fabric API (sometimes even to zero).

Reference#

pydantic model bfabric.experimental.workunit_definition.WorkunitDefinition#

Bases: BaseModel

Defines a workunit, including details on how to execute it and where to register it. This class provides a simple way for developers to persist and run workunit definitions from YAML files, as well as loading the same from B-Fabric workunits. This abstraction ensures easier development and testing of applications.

Show JSON schema

{
   "title": "WorkunitDefinition",
   "description": "Defines a workunit, including details on how to execute it and where to register it.\nThis class provides a simple way for developers to persist and run workunit definitions from YAML files, as well as\nloading the same from B-Fabric workunits. This abstraction ensures easier development and testing of applications.",
   "type": "object",
   "properties": {
      "execution": {
         "$ref": "#/$defs/WorkunitExecutionDefinition"
      },
      "registration": {
         "anyOf": [
            {
               "$ref": "#/$defs/WorkunitRegistrationDefinition"
            },
            {
               "type": "null"
            }
         ]
      }
   },
   "$defs": {
      "WorkunitExecutionDefinition": {
         "description": "Defines the execution details of a workunit, i.e. the inputs necessary to compute the results,\nbut not the final details on how to register the results in B-Fabric.",
         "properties": {
            "raw_parameters": {
               "additionalProperties": {
                  "anyOf": [
                     {
                        "type": "string"
                     },
                     {
                        "type": "null"
                     }
                  ]
               },
               "title": "Raw Parameters",
               "type": "object"
            },
            "dataset": {
               "anyOf": [
                  {
                     "type": "integer"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Dataset"
            },
            "resources": {
               "default": [],
               "items": {
                  "type": "integer"
               },
               "title": "Resources",
               "type": "array"
            }
         },
         "required": [
            "raw_parameters"
         ],
         "title": "WorkunitExecutionDefinition",
         "type": "object"
      },
      "WorkunitRegistrationDefinition": {
         "description": "Defines the B-Fabric registration details of a workunit.",
         "properties": {
            "application_id": {
               "title": "Application Id",
               "type": "integer"
            },
            "application_name": {
               "title": "Application Name",
               "type": "string"
            },
            "workunit_id": {
               "title": "Workunit Id",
               "type": "integer"
            },
            "workunit_name": {
               "title": "Workunit Name",
               "type": "string"
            },
            "container_id": {
               "title": "Container Id",
               "type": "integer"
            },
            "container_type": {
               "enum": [
                  "project",
                  "order"
               ],
               "title": "Container Type",
               "type": "string"
            },
            "storage_id": {
               "title": "Storage Id",
               "type": "integer"
            },
            "storage_output_folder": {
               "format": "path",
               "title": "Storage Output Folder",
               "type": "string"
            }
         },
         "required": [
            "application_id",
            "application_name",
            "workunit_id",
            "workunit_name",
            "container_id",
            "container_type",
            "storage_id",
            "storage_output_folder"
         ],
         "title": "WorkunitRegistrationDefinition",
         "type": "object"
      }
   },
   "required": [
      "execution",
      "registration"
   ]
}

Fields:

execution (bfabric.experimental.workunit_definition.WorkunitExecutionDefinition)
registration (bfabric.experimental.workunit_definition.WorkunitRegistrationDefinition | None)

field execution: WorkunitExecutionDefinition [Required]#: Execution details of the workunit.

field registration: WorkunitRegistrationDefinition | None [Required]#: Registration details of the workunit.

classmethod from_ref(workunit: Path | int, client: Bfabric, cache_file: Path | None = None) → WorkunitDefinition#

Loads the workunit definition from the provided reference, which can be a path to a YAML file, or a workunit ID.

If the cache file is provided and exists, it will be loaded directly instead of resolving the reference. Otherwise, the result will be cached to the provided file. :param workunit: The workunit reference, which can be a path to a YAML file, or a workunit ID. :param client: The B-Fabric client to use for resolving the workunit. :param cache_file: The path to the cache file, if any.

classmethod from_workunit(workunit: Workunit) → WorkunitDefinition#: Loads the workunit definition from the provided B-Fabric workunit.

classmethod from_yaml(path: Path) → WorkunitDefinition#: Loads the workunit definition from the provided path.

to_yaml(path: Path) → None#: Writes the workunit definition to the provided path.

pydantic model bfabric.experimental.workunit_definition.WorkunitExecutionDefinition#

Bases: BaseModel

Defines the execution details of a workunit, i.e. the inputs necessary to compute the results, but not the final details on how to register the results in B-Fabric.

Show JSON schema

{
   "title": "WorkunitExecutionDefinition",
   "description": "Defines the execution details of a workunit, i.e. the inputs necessary to compute the results,\nbut not the final details on how to register the results in B-Fabric.",
   "type": "object",
   "properties": {
      "raw_parameters": {
         "additionalProperties": {
            "anyOf": [
               {
                  "type": "string"
               },
               {
                  "type": "null"
               }
            ]
         },
         "title": "Raw Parameters",
         "type": "object"
      },
      "dataset": {
         "anyOf": [
            {
               "type": "integer"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Dataset"
      },
      "resources": {
         "default": [],
         "items": {
            "type": "integer"
         },
         "title": "Resources",
         "type": "array"
      }
   },
   "required": [
      "raw_parameters"
   ]
}

Fields:

dataset (int | None)
raw_parameters (dict[str, str | None])
resources (list[int])

Validators:

either_dataset_or_resources » all fields
mutually_exclusive_dataset_resources » all fields

field dataset: int | None = None#

Input dataset (for dataset-flow applications)

Validated by:

either_dataset_or_resources
mutually_exclusive_dataset_resources

field raw_parameters: dict[str, str | None] [Required]#

The parameters passed to the workunit, in their raw form, i.e. everything is a string or None.

Validated by:

either_dataset_or_resources
mutually_exclusive_dataset_resources

field resources: list[int] = []#

Input resources (for resource-flow applications

Validated by:

either_dataset_or_resources
mutually_exclusive_dataset_resources

classmethod from_workunit(workunit: Workunit) → WorkunitExecutionDefinition#: Loads the workunit execution definition from the provided B-Fabric workunit.

validator either_dataset_or_resources » all fields#: Validates that either dataset or resources are provided.

validator mutually_exclusive_dataset_resources » all fields#: Validates that dataset and resources are mutually exclusive.

pydantic model bfabric.experimental.workunit_definition.WorkunitRegistrationDefinition#

Bases: BaseModel

Defines the B-Fabric registration details of a workunit.

Show JSON schema

{
   "title": "WorkunitRegistrationDefinition",
   "description": "Defines the B-Fabric registration details of a workunit.",
   "type": "object",
   "properties": {
      "application_id": {
         "title": "Application Id",
         "type": "integer"
      },
      "application_name": {
         "title": "Application Name",
         "type": "string"
      },
      "workunit_id": {
         "title": "Workunit Id",
         "type": "integer"
      },
      "workunit_name": {
         "title": "Workunit Name",
         "type": "string"
      },
      "container_id": {
         "title": "Container Id",
         "type": "integer"
      },
      "container_type": {
         "enum": [
            "project",
            "order"
         ],
         "title": "Container Type",
         "type": "string"
      },
      "storage_id": {
         "title": "Storage Id",
         "type": "integer"
      },
      "storage_output_folder": {
         "format": "path",
         "title": "Storage Output Folder",
         "type": "string"
      }
   },
   "required": [
      "application_id",
      "application_name",
      "workunit_id",
      "workunit_name",
      "container_id",
      "container_type",
      "storage_id",
      "storage_output_folder"
   ]
}

Fields:

application_id (int)
application_name (str)
container_id (int)
container_type (Literal['project', 'order'])
storage_id (int)
storage_output_folder (pathlib.Path)
workunit_id (int)
workunit_name (str)

field application_id: int [Required]#: The ID of the executing application.

field application_name: PathSafeStr [Required]#

The name of the executing application.

Constraints:

func = <function path_safe_name at 0x1138f09a0>
json_schema_input_type = PydanticUndefined

field container_id: int [Required]#: The ID of the container.

field container_type: Literal['project', 'order'] [Required]#: The type of the container.

field storage_id: int [Required]#: The ID of the storage.

field storage_output_folder: Path [Required]#: The output folder in the storage.

field workunit_id: int [Required]#: The ID of the workunit.

field workunit_name: PathSafeStr [Required]#

The name of the workunit.

Constraints:

func = <function path_safe_name at 0x1138f09a0>
json_schema_input_type = PydanticUndefined

classmethod from_workunit(workunit: Workunit) → WorkunitRegistrationDefinition#: Loads the workunit registration definition from the provided B-Fabric workunit.

Workunit Definition

Contents

Workunit Definition#

Creating WorkunitDefinition instances#

Workunit references#

Reference#