pynenc.workflow.workflow_context

Module Contents

Classes

WorkflowContext

Provides workflow capabilities for tasks within Pynenc.

Data

S

API

pynenc.workflow.workflow_context.S

‘TypeVar(…)’

class pynenc.workflow.workflow_context.WorkflowContext(task: pynenc.task.Task)

Provides workflow capabilities for tasks within Pynenc.

This class exposes methods for workflow state management, determinism, and durability. It’s accessible via the wf property on any Pynenc task.

Parameters:

task – The task this helper is attached to

Initialization

Initialize the workflow helper with its associated task.

property deterministic: pynenc.workflow.workflow_deterministic.DeterministicExecutor

Get the deterministic executor for this workflow context.

property app: pynenc.app.Pynenc

Get the Pynenc app instance associated with this workflow.

property identity: pynenc.workflow.workflow_identity.WorkflowIdentity

Get the workflow identity for the current execution.

get_data(key: str, default: Any = None) → Any

Get a value from the workflow’s data store.

Any task within the workflow can read data.

Parameters:

• key – The data key to retrieve

• default – Default value if the key doesn’t exist

Returns:

The stored value or default

set_data(key: str, value: Any) → None

Set a value in the workflow’s data store.

Only the main workflow task can write workflow data.

Parameters:

• key – The data key to set

• value – The value to store

Raises:

WorkflowAccessError – If a non-owner task attempts to modify data

random() → float

Generate a deterministic random number in the workflow context.

Returns the same sequence of “random” numbers on workflow replay.

Returns:

A random float between 0.0 and 1.0

utc_now() → datetime.datetime

Get the current time deterministically in the workflow context.

Returns deterministic timestamps that advance consistently on workflow replay.

Returns:

Current datetime with UTC timezone

uuid() → str

Generate a deterministic UUID in the workflow context.

Returns the same sequence of UUIDs on workflow replay.

Returns:

UUID string

execute_task(task: pynenc.task.Task, *args: Any, **kwargs: Any) → Any

Execute a task as part of this workflow with deterministic replay.

On workflow replay, returns the recorded result without re-execution.

Parameters:

• task – The task to execute

• args – Positional arguments for the task

• kwargs – Keyword arguments for the task

Returns:

Task result