pynenc.invocation.status

Invocation status management.

This module defines the status lifecycle of task invocations, including state transitions, ownership rules, and the state machine that enforces them.

Key components:

  • InvocationStatus: Enum of all possible invocation states

  • InvocationStatusRecord: Immutable record combining status with ownership

  • StatusDefinition: Declarative rules for status behavior

  • State machine functions: Validation and transition logic

Module Contents

Classes

InvocationStatus

An enumeration representing the status of a task invocation.

InvocationStatusRecord

Immutable record combining status with ownership information.

StatusDefinition

Declarative definition of status behavior and ownership rules.

StatusConfiguration

Complete configuration for invocation status behavior.

Functions

get_status_definition

Get the status definition for a given invocation status.

validate_transition

Validate state transition or raise exception.

validate_ownership

Validate ownership requirements for a transition.

compute_new_owner

Compute new owner based on status transition.

status_record_transition

Execute a status change with safety checks.

Data

_CONFIG

API

class pynenc.invocation.status.InvocationStatus[source]

Bases: enum.StrEnum

An enumeration representing the status of a task invocation.

The PENDING status will expire after the time specified in

Attr:

~pynenc.conf.config_pynenc.ConfigPynenc.max_pending_seconds.

Note

SUCCESS, FAILED, and CONCURRENCY_CONTROLLED_FINAL are final statuses that terminate the invocation lifecycle.

Variables:

  • REGISTERED – The task call has been routed and is registered

  • CONCURRENCY_CONTROLLED – The task call is not allowed to run due to concurrency control

  • CONCURRENCY_CONTROLLED_FINAL – The task call was blocked by concurrency control and will not be retried

  • REROUTED – The task call has been re-routed and is registered

  • PENDING – The task call was picked by a runner but is not yet executed

  • PENDING_RECOVERY – The task call exceeded PENDING timeout and is being recovered

  • RUNNING – The task call is currently running

  • RUNNING_RECOVERY – The task call is being recovered because the owner runner is inactive

  • PAUSED – The task call execution is paused

  • RESUMED – The task call execution has been resumed

  • KILLED – The task call execution has been killed

  • SUCCESS – The task call finished without errors

  • FAILED – The task call finished with exceptions

  • RETRY – The task call finished with a retriable exception

Initialization

Initialize self. See help(type(self)) for accurate signature.

REGISTERED

‘registered’

CONCURRENCY_CONTROLLED

‘concurrency_controlled’

CONCURRENCY_CONTROLLED_FINAL

‘concurrency_controlled_final’

REROUTED

‘rerouted’

PENDING

‘pending’

PENDING_RECOVERY

‘pending_recovery’

RUNNING

‘running’

RUNNING_RECOVERY

‘running_recovery’

PAUSED

‘paused’

RESUMED

‘resumed’

KILLED

‘killed’

SUCCESS

‘success’

FAILED

‘failed’

RETRY

‘retry’

is_final() → bool[source]

Check if the status terminates the invocation lifecycle.

is_available_for_run() → bool[source]

Check if the task can be picked up and run by any broker.

can_transition_to(target: pynenc.invocation.status.InvocationStatus) → bool[source]

Check if this status has a valid transition to the target status.

classmethod get_final_statuses() → frozenset[pynenc.invocation.status.InvocationStatus][source]

Return all statuses that terminate the invocation lifecycle.

classmethod get_available_for_run_statuses() → frozenset[pynenc.invocation.status.InvocationStatus][source]

Return all statuses where invocations can be picked up by runners.

class pynenc.invocation.status.InvocationStatusRecord[source]

Immutable record combining status with ownership information.

Variables:

  • status (InvocationStatus) – The current invocation status

  • runner_id (str | None) – Runner ID that owns this invocation (None if no owner)

  • timestamp (datetime) – When this status was set

status: pynenc.invocation.status.InvocationStatus

None

runner_id: str | None

None

timestamp: datetime.datetime

‘field(…)’

to_json() → dict[source]

Serialize the status record to a JSON-compatible dictionary.

classmethod from_json(json_dict: dict) → pynenc.invocation.status.InvocationStatusRecord[source]

Deserialize a JSON-compatible dictionary into a status record.

class pynenc.invocation.status.StatusDefinition[source]

Declarative definition of status behavior and ownership rules.

Variables:

  • allowed_transitions (frozenset[InvocationStatus]) – Valid next statuses

  • is_final (bool) – Terminates invocation lifecycle

  • available_for_run (bool) – Can be picked up by runners

  • requires_ownership (bool) – Only owner can modify

  • acquires_ownership (bool) – Claims ownership on entry

  • releases_ownership (bool) – Releases ownership on entry

  • overrides_ownership (bool) – Bypasses ownership validation (for recovery scenarios)

allowed_transitions: frozenset[pynenc.invocation.status.InvocationStatus]

‘field(…)’

is_final: bool

False

available_for_run: bool

False

requires_ownership: bool

False

acquires_ownership: bool

False

releases_ownership: bool

False

overrides_ownership: bool

False

__post_init__() → None[source]
class pynenc.invocation.status.StatusConfiguration[source]

Complete configuration for invocation status behavior.

Variables:

definitions (dict[InvocationStatus | None, StatusDefinition]) – Behavior rules per status

definitions: dict[pynenc.invocation.status.InvocationStatus | None, pynenc.invocation.status.StatusDefinition]

None

__post_init__() → None[source]
final_statuses() → frozenset[pynenc.invocation.status.InvocationStatus]
available_for_run_statuses() → frozenset[pynenc.invocation.status.InvocationStatus]
ownership_required_statuses() → frozenset[pynenc.invocation.status.InvocationStatus]
ownership_acquire_statuses() → frozenset[pynenc.invocation.status.InvocationStatus]
ownership_release_statuses() → frozenset[pynenc.invocation.status.InvocationStatus]
get_definition(status: pynenc.invocation.status.InvocationStatus | None) → pynenc.invocation.status.StatusDefinition[source]
pynenc.invocation.status._CONFIG: Final[pynenc.invocation.status.StatusConfiguration]

‘StatusConfiguration(…)’

pynenc.invocation.status.get_status_definition(status: pynenc.invocation.status.InvocationStatus) → pynenc.invocation.status.StatusDefinition[source]

Get the status definition for a given invocation status.

Parameters:

status – The invocation status to look up

Returns:

The status definition with rules and behavior

pynenc.invocation.status.validate_transition(from_status: pynenc.invocation.status.InvocationStatus | None, to_status: pynenc.invocation.status.InvocationStatus) → None[source]

Validate state transition or raise exception.

Parameters:
Raises:

InvocationStatusTransitionError – If transition is invalid

pynenc.invocation.status.validate_ownership(current_record: pynenc.invocation.status.InvocationStatusRecord | None, new_status: pynenc.invocation.status.InvocationStatus, runner_id: str | None) → None[source]

Validate ownership requirements for a transition.

Parameters:
Raises:

InvocationStatusOwnershipError – If ownership rules are violated

pynenc.invocation.status.compute_new_owner(current_record: pynenc.invocation.status.InvocationStatusRecord | None, new_status: pynenc.invocation.status.InvocationStatus, runner_id: str | None) → str | None[source]

Compute new owner based on status transition.

pynenc.invocation.status.status_record_transition(current_record: pynenc.invocation.status.InvocationStatusRecord | None, new_status: pynenc.invocation.status.InvocationStatus, runner_id: str | None) → pynenc.invocation.status.InvocationStatusRecord[source]

Execute a status change with safety checks.

Parameters:
Returns:

New validated status record

Raises: