pynenc.state_backend.redis_state_backend

Module Contents

Classes

RedisStateBackend

A Redis-based implementation of the state backend.

API

class pynenc.state_backend.redis_state_backend.RedisStateBackend(app: pynenc.app.Pynenc)

Bases: pynenc.state_backend.base_state_backend.BaseStateBackend

A Redis-based implementation of the state backend.

This backend uses Redis to store and retrieve the state of invocations, including their data, history, results, and exceptions. It’s suitable for distributed systems where shared state management is required.

Initialization

conf() → pynenc.conf.config_state_backend.ConfigStateBackendRedis

property client: redis.Redis

Lazy initialization of Redis client

purge() → None

Clears all data from the Redis backend for the current app.app_id.

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

Inserts or updates an invocation in Redis.

Parameters:

invocation (DistributedInvocation) – The invocation object to upsert.

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

Retrieves an invocation from Redis by its ID.

Parameters:

invocation_id (DistributedInvocation) – 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 in Redis.

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 from Redis.

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 in Redis.

Parameters:

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

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

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

Retrieves the result of an invocation from Redis.

Parameters:

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

Returns:

The result of the invocation.

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

Sets the exception for an invocation in Redis.

Parameters:

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

• exception (Exception) – The exception to set.

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

Retrieves the exception of an invocation from Redis.

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 (must be serializable)

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]

Retrieve all app information registered in this state backend.

Returns:

Dictionary mapping app_id to app information