pynenc.workflow.workflow_context

Module Contents

Classes

WorkflowContext	Provides workflow capabilities for tasks within Pynenc.
WorkflowRootOps	Root-only workflow orchestration operations.
WorkflowRootContext	Workflow context for workflow-defining tasks.

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 shared workflow state for any task that is running inside a workflow. Root-only orchestration lives on WorkflowRootContext.root.

Parameters:

task – The task this helper is attached to

Initialization

Initialize the workflow helper with its associated task.

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.

Any task within the workflow can write data.

Parameters:

• key – The data key to set

• value – The value to store

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

Root-only workflow orchestration operations.

The public wf.root boundary makes it explicit that these methods are only valid on the workflow-defining invocation. The deterministic replay cursor itself is stored on the active invocation object, not on the cached task-level workflow context.

Initialization

property _invocation: pynenc.invocation.dist_invocation.DistributedInvocation

_raise_scope_error(operation_name: str) → NoReturn

_ensure_root_workflow_operation_allowed(operation_name: str) → pynenc.invocation.dist_invocation.DistributedInvocation

_deterministic(operation_name: str) → pynenc.workflow.workflow_deterministic.DeterministicExecutor

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

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

Bases: pynenc.workflow.workflow_context.WorkflowContext

Workflow context for workflow-defining tasks.

Initialization

Initialize the workflow helper with its associated task.