`Workflow` model reference¶

Experimental feature

This feature is experimental and should be considered unstable and unsupported. API may change at any time and without backwards compatibility.

sheppy.experimental.Workflow ¶

Bases: BaseModel

A workflow instance created when a workflow function is called.

A workflow is a durable generator: each yield of a Task (or a list of tasks for fan-out) is an await point. Yielded tasks are queued, and once they reach a terminal state the generator is replayed from the start with finished tasks fed back in, until it either returns (workflow completed) or raises (workflow failed).

ATTRIBUTE	DESCRIPTION
`id`	Unique identifier for the workflow. TYPE: `UUID`
`func`	Fully qualified function name, e.g. `my_module:my_workflow`. TYPE: `str`
`args`	Positional arguments passed to the workflow function. TYPE: `tuple[Any, ...]`
`kwargs`	Keyword arguments passed to the workflow function. TYPE: `dict[str, Any]`
`task_order`	Deterministic task IDs in creation order (one per replay step). TYPE: `list[str]`
`completed`	True when the workflow function has returned. TYPE: `bool`
`final_result`	Return value of the workflow function (when completed). TYPE: `Any`
`exception`	Exception data if the workflow failed, as a TaskException. None otherwise. TYPE: `TaskException \| None`
`error`	Deprecated, use `exception` instead. TYPE: `str \| None`
`created_at`	Timestamp when the workflow was created. TYPE: `AwareDatetime`
`finished_at`	Timestamp when the workflow completed or failed. None while running. TYPE: `AwareDatetime \| None`

Note

You should not create Workflow instances directly. Instead, use the @workflow decorator to define a workflow function, and then call that function to create a Workflow instance.
args and kwargs must be JSON serializable.
The workflow function must be deterministic: the sequence of tasks it creates may only depend on the results of previously yielded tasks, never on external state or randomness.

Example

from sheppy import task
from sheppy.experimental import workflow

@task
async def add(x: int, y: int) -> int:
    return x + y

@workflow
def my_workflow(x: int):
    a = yield add(x, 1)          # a is the finished Task, a.result == x + 1
    b = yield add(a.result, 10)  # sequential step
    return b.result

wf = my_workflow(5)  # returns a Workflow instance, it is NOT executed yet

id ¶

id: UUID = Field(default_factory=uuid4)

UUID: Unique identifier for the workflow.

func ¶

func: str

str: Fully qualified function name, e.g. my_module:my_workflow

args ¶

args: tuple[Any, ...] = Field(default_factory=tuple)

tuple[Any, ...]: Positional arguments passed to the workflow function.

kwargs ¶

kwargs: dict[str, Any] = Field(default_factory=dict)

dict[str, Any]: Keyword arguments passed to the workflow function.

ttl ¶

ttl: TTLValue = 'inherit'

Default metadata TTL applied to workflow tasks that inherit it.

error_ttl ¶

error_ttl: TTLValue = 'inherit'

Default metadata error TTL applied to workflow tasks that inherit it.

task_order ¶

task_order: list[str] = Field(default_factory=list)

list[str]: Deterministic task IDs in creation order (one per replay step).

completed ¶

completed: bool = False

bool: True when the workflow function has returned.

final_result ¶

final_result: Any = None

Any: Return value of the workflow function (when completed).

exception ¶

exception: TaskException | None = None

TaskException|None: Exception data if the workflow failed. None otherwise.

created_at ¶

created_at: AwareDatetime = Field(
    default_factory=lambda: datetime.now(timezone.utc)
)

datetime: Timestamp when the workflow was created.

finished_at ¶

finished_at: AwareDatetime | None = None

datetime|None: Timestamp when the workflow completed or failed. None while running.

error ¶

error: str | None

Deprecated, use workflow.exception instead. Returns the error as a "Type: message" string.

get_stored_task_ids ¶

get_stored_task_ids() -> dict[int, UUID]

Map of replay step index to deterministic task ID, from previous runs.

Source code in src/sheppy/experimental/_workflow.py

def get_stored_task_ids(self) -> dict[int, UUID]:
    """Map of replay step index to deterministic task ID, from previous runs."""
    return {idx: UUID(tid) for idx, tid in enumerate(self.task_order)}

Workflow model reference¶