pynenc.state_backend.mem_state_backend

Module Contents

Classes

MemStateBackend

A memory-based implementation of the state backend.

Data

_APP_INFO_REGISTRY

API

pynenc.state_backend.mem_state_backend._APP_INFO_REGISTRY: dict[str, pynenc.app.AppInfo]

None

class pynenc.state_backend.mem_state_backend.MemStateBackend(app: pynenc.app.Pynenc)

Bases: pynenc.state_backend.base_state_backend.BaseStateBackend

A memory-based implementation of the state backend.

Stores invocation data, history, results, and exceptions in in-memory dictionaries. Useful for environments where persistence is not required or for testing purposes.

Warning

The MemStateBackend class stores all data in the process’s memory and is not suitable for production systems. Its use should be limited to testing or demonstration purposes only.

Initialization

purge() → None

Clears all stored data

_upsert_invocation(invocation: pynenc.invocation.dist_invocation.DistributedInvocation) → None

Inserts or updates an invocation in the memory cache.

Parameters:

invocation (DistributedInvocation) – The invocation object to upsert.

_get_invocation(invocation_id: str) → DistributedInvocation[Params, Result]

Retrieves an invocation from the memory cache by its ID.

Parameters:

invocation_id (str) – The ID of the invocation to retrieve.

Returns:

The retrieved invocation object.

_add_history(invocation: pynenc.invocation.dist_invocation.DistributedInvocation, invocation_history: pynenc.state_backend.base_state_backend.InvocationHistory) → None

Adds a history record to an invocation.

Parameters:

• invocation (DistributedInvocation) – The invocation to add history for.

• invocation_history (InvocationHistory) – The history record to add.

_get_history(invocation: DistributedInvocation[Params, Result]) → list[pynenc.state_backend.base_state_backend.InvocationHistory]

Retrieves the history of an invocation.

Parameters:

invocation (DistributedInvocation) – The invocation to get the history for.

Returns:

A list of invocation history records.

_set_result(invocation: DistributedInvocation[Params, Result], result: pynenc.types.Result) → None

Sets the result for an invocation.

Parameters:

• invocation (DistributedInvocation) – The invocation to set the result for.

• result (Result) – The result of the invocation.

_set_exception(invocation: DistributedInvocation[Params, Result], exception: Exception) → None

Sets the exception for an invocation.

Parameters:

• invocation (DistributedInvocation) – The invocation to set the exception for.

• exception (Exception) – The exception to set.

_get_result(invocation: DistributedInvocation[Params, Result]) → pynenc.types.Result

Retrieves the result of an invocation.

Parameters:

invocation (DistributedInvocation) – The invocation to get the result for.

Returns:

The result of the invocation.

_get_exception(invocation: DistributedInvocation[Params, Result]) → Exception

Retrieves the exception of an invocation.

Parameters:

invocation (DistributedInvocation) – The invocation to get the exception for.

Returns:

The exception of the invocation.

get_workflow_data(workflow_identity: pynenc.workflow.WorkflowIdentity, key: str, default: Any = None) → Any

Get a value from workflow data.

Parameters:

• workflow_identity – Workflow identity

• key – Data key to retrieve

• default – Default value if key doesn’t exist

Returns:

Stored value or default

set_workflow_data(workflow_identity: pynenc.workflow.WorkflowIdentity, key: str, value: Any) → None

Set a value in workflow data.

Parameters:

• workflow_identity – Workflow identity

• key – Data key to set

• value – Value to store

get_workflow_deterministic_value(workflow: pynenc.workflow.WorkflowIdentity, key: str) → Any

Retrieve a deterministic value for workflow operations.

Parameters:

• workflow – The workflow identity

• key – Key identifying the deterministic value

Returns:

The stored value or None if not found

set_workflow_deterministic_value(workflow: pynenc.workflow.WorkflowIdentity, key: str, value: Any) → None

Store a deterministic value for workflow operations.

Parameters:

• workflow – The workflow identity

• key – Key identifying the deterministic value

• value – The value to store

store_app_info(app_info: pynenc.app.AppInfo) → None

Register this app’s information in the state backend for discovery.

Parameters:

app_info – The app information to store

get_app_info() → pynenc.app.AppInfo

Retrieve information of the current app.

Returns:

The app information

Raises:

ValueError – If app info is not found

static get_all_app_infos() → dict[str, pynenc.app.AppInfo]

store_workflow_run(workflow_identity: pynenc.workflow.WorkflowIdentity) → None

Store a workflow run for tracking and monitoring.

Parameters:

workflow_identity – The workflow identity to store

get_all_workflows() → Iterator[str]

Retrieve all workflow types (workflow_task_ids) stored in this state backend.

Returns:

Iterator of workflow task IDs representing different workflow types

get_all_workflows_runs() → Iterator[pynenc.workflow.WorkflowIdentity]

Retrieve workflow run identities from this state backend.

Returns:

Iterator of workflow identities for runs

get_workflow_runs(workflow_task_id: str) → Iterator[pynenc.workflow.WorkflowIdentity]

Retrieve workflow run identities from this state backend.

Parameters:

workflow_task_id – Filter for specific workflow type

Returns:

Iterator of workflow identities for runs

store_workflow_sub_invocation(parent_workflow_id: str, sub_invocation_id: str) → None

Store a sub-invocation ID that runs inside a parent workflow.

Parameters:

• parent_workflow_id – The workflow ID that contains the sub-invocation

• sub_invocation_id – The invocation ID of the task/sub-workflow running inside

get_workflow_sub_invocations(workflow_id: str) → Iterator[str]

Retrieve all sub-invocation IDs that run inside a specific workflow.

Parameters:

workflow_id – The workflow ID to get sub-invocations for

Returns:

Iterator of invocation IDs that run inside the workflow