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

class bfabric.experimental.workunit_definition.WorkunitDefinition(*, execution: WorkunitExecutionDefinition, registration: WorkunitRegistrationDefinition | None)

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.

execution: WorkunitExecutionDefinition

Execution 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.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

registration: WorkunitRegistrationDefinition | None

Registration details of the workunit.

to_yaml(path: Path) → None

Writes the workunit definition to the provided path.

class bfabric.experimental.workunit_definition.WorkunitExecutionDefinition(*, raw_parameters: dict[str, str | None], dataset: int | None = None, resources: list[int] = [])

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.

dataset: int | None

Input dataset (for dataset-flow applications)

either_dataset_or_resources() → WorkunitExecutionDefinition

Validates that either dataset or resources are provided.

classmethod from_workunit(workunit: Workunit) → WorkunitExecutionDefinition

Loads the workunit execution definition from the provided B-Fabric workunit.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

mutually_exclusive_dataset_resources() → WorkunitExecutionDefinition

Validates that dataset and resources are mutually exclusive.

raw_parameters: dict[str, str | None]

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

resources: list[int]

Input resources (for resource-flow applications

class bfabric.experimental.workunit_definition.WorkunitRegistrationDefinition(*, application_id: int, application_name: Annotated[str, BeforeValidator(func=path_safe_name, json_schema_input_type=PydanticUndefined)], workunit_id: int, workunit_name: Annotated[str, BeforeValidator(func=path_safe_name, json_schema_input_type=PydanticUndefined)], container_id: int, container_type: Literal['project', 'order'], storage_id: int, storage_output_folder: Path, user_id: int | None = None)

Bases: BaseModel

Defines the B-Fabric registration details of a workunit.

application_id: int

The ID of the executing application.

application_name: PathSafeStr

The name of the executing application.

container_id: int

The ID of the container.

container_type: Literal['project', 'order']

The type of the container.

classmethod from_workunit(workunit: Workunit) → WorkunitRegistrationDefinition

Loads the workunit registration definition from the provided B-Fabric workunit.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

storage_id: int

The ID of the storage.

storage_output_folder: Path

The output folder in the storage.

user_id: int | None

The ID of the user who created the workunit, if available.

workunit_id: int

The ID of the workunit.

workunit_name: PathSafeStr

The name of the workunit.