pynenc.exceptions

Global Pynenc exception and warning classes.

Module Contents

API

exception pynenc.exceptions.PynencError

Bases: Exception

Base class for all Pynenc related errors.

Initialization

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

_to_json_dict() → dict[str, Any]

Returns:

a json serializable dictionary

classmethod _from_json_dict(json_dict: dict[str, Any]) → pynenc.exceptions.PynencError

Returns:

a new error from the serialized json compatible dictionary

to_json() → str

Returns:

the serialized error

classmethod from_json(error_name: str, serialized: str) → pynenc.exceptions.PynencError

Returns:

the child class from a serialized error

exception pynenc.exceptions.RetryError

Bases: pynenc.exceptions.PynencError

Error raised when a task should be retried.

Initialization

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

exception pynenc.exceptions.ConcurrencyRetryError

Bases: pynenc.exceptions.RetryError

Error raised when a task should be retried due to concurrency control.

Initialization

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

exception pynenc.exceptions.SerializationError(message: str)

Bases: pynenc.exceptions.PynencError

Error raised when an argument cannot be serialized.

Wraps the original serialization error with context about which argument failed and truncated value information to aid debugging.

Initialization

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

__str__() → str

exception pynenc.exceptions.TaskError(task_id: pynenc.identifiers.task_id.TaskId, message: str | None = None)

Bases: pynenc.exceptions.PynencError

Base class for all Task related errors.

Initialization

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

__str__() → str

_to_json_dict() → dict[str, Any]

Returns:

the serialized error

classmethod _from_json_dict(json_dict: dict[str, Any]) → pynenc.exceptions.TaskError

Returns:

the serialized error

exception pynenc.exceptions.InvalidTaskOptionsError(task_id: pynenc.identifiers.task_id.TaskId, message: str | None = None)

Bases: pynenc.exceptions.TaskError

Error raised when the task options are invalid.

Initialization

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

exception pynenc.exceptions.TaskRoutingError(task_id: pynenc.identifiers.task_id.TaskId, message: str | None = None)

Bases: pynenc.exceptions.TaskError

Error raised when a task will not be routed.

Initialization

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

exception pynenc.exceptions.TaskParallelProcessingError(task_id: pynenc.identifiers.task_id.TaskId, message: str | None = None)

Bases: pynenc.exceptions.TaskError

Error parallelizing a task.

Initialization

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

exception pynenc.exceptions.InvocationConcurrencyWithDifferentArgumentsError(task_id: pynenc.identifiers.task_id.TaskId, existing_invocation_id: str, new_call_id: pynenc.identifiers.call_id.CallId, diff: str, message: str | None = None)

Bases: pynenc.exceptions.TaskRoutingError

Error raised when there is a task with different arguments than the current task.

Initialization

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

classmethod from_call_mismatch(existing_invocation: pynenc.invocation.BaseInvocation, new_call: pynenc.call.Call, message: str | None = None) → pynenc.exceptions.InvocationConcurrencyWithDifferentArgumentsError

static format_difference(existing_call: pynenc.call.Call, new_call: pynenc.call.Call) → str

Format argument differences in a clear, concise way.

__str__() → str

_to_json_dict() → dict[str, Any]

Returns:

the serialized error

classmethod _from_json_dict(json_dict: dict[str, str]) → pynenc.exceptions.InvocationConcurrencyWithDifferentArgumentsError

Returns:

a new error from a serialized error

exception pynenc.exceptions.InvocationError(invocation_id: str, message: str | None = None)

Bases: pynenc.exceptions.PynencError

Base class for all Task related errors.

Initialization

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

__str__() → str

_to_json_dict() → dict[str, Any]

classmethod _from_json_dict(json_dict: dict[str, Any]) → pynenc.exceptions.InvocationError

exception pynenc.exceptions.StateBackendError

Bases: pynenc.exceptions.PynencError

Error raised when a task will not be routed.

Initialization

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

exception pynenc.exceptions.InvocationNotFoundError(invocation_id: str, message: str | None = None)

Bases: pynenc.exceptions.StateBackendError

Error raised when the invocation is not present in the State Backend.

Initialization

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

__str__() → str

_to_json_dict() → dict[str, Any]

classmethod _from_json_dict(json_dict: dict[str, Any]) → pynenc.exceptions.InvocationNotFoundError

exception pynenc.exceptions.RunnerNotExecutableError

Bases: pynenc.exceptions.PynencError

Raised when trying to execute a runner that is not meant to be executed.

Initialization

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

exception pynenc.exceptions.RunnerError

Bases: pynenc.exceptions.PynencError

Base class for all Runner related errors.

Initialization

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

exception pynenc.exceptions.ConfigError

Bases: pynenc.exceptions.PynencError

Base class for all the config related errors

Initialization

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

exception pynenc.exceptions.ConfigMultiInheritanceError

Bases: pynenc.exceptions.ConfigError

Error related with multiinheritance of config fields

Initialization

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

exception pynenc.exceptions.AlreadyInitializedError

Bases: pynenc.exceptions.PynencError

Error raised when trying to change the class of a component after it was initialized

Initialization

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

exception pynenc.exceptions.InvocationStatusError

Bases: pynenc.exceptions.PynencError

Base class for all invocation status related errors.

Use this to catch any error related to status transitions, ownership, or final status modifications.

Initialization

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

exception pynenc.exceptions.InvocationStatusRaceConditionError(invocation_id: str, previous_status_record: pynenc.invocation.InvocationStatusRecord, expected_status_record: pynenc.invocation.InvocationStatusRecord, actual_status_record: pynenc.invocation.InvocationStatusRecord)

Bases: pynenc.exceptions.InvocationStatusError

Raised when a race condition is detected in non-atomic status updates.

This error occurs in non-atomic orchestrators when the actual status differs from the expected status after a write operation, indicating concurrent modification.

Initialization

Create a race condition error.

Parameters:

• invocation_id (str) – The invocation that experienced the race condition

• previous_status_record (InvocationStatusRecord) – Status before the attempted change

• expected_status_record (InvocationStatusRecord) – Status that was intended

• actual_status_record (InvocationStatusRecord) – Status that was actually set

__str__() → str

_to_json_dict() → dict[str, Any]

classmethod _from_json_dict(json_dict: dict[str, Any]) → pynenc.exceptions.InvocationStatusRaceConditionError

exception pynenc.exceptions.InvocationStatusTransitionError(from_status: InvocationStatus | None, to_status: pynenc.invocation.InvocationStatus, allowed_statuses: frozenset[pynenc.invocation.InvocationStatus])

Bases: pynenc.exceptions.InvocationStatusError

Raised when attempting an invalid invocation status transition.

This error occurs when trying to change an invocation’s status to a state that is not allowed by the state machine rules.

Initialization

Create an invalid state transition error.

Parameters:

• from_status (InvocationStatus | None) – Current status

• to_status (InvocationStatus) – Attempted target status

• allowed_statuses (frozenset[InvocationStatus]) – Statuses that are allowed from current state

__str__() → str

_to_json_dict() → dict[str, Any]

classmethod _from_json_dict(json_dict: dict[str, Any]) → pynenc.exceptions.InvocationStatusTransitionError

exception pynenc.exceptions.InvocationStatusOwnershipError(from_status: InvocationStatus | None, to_status: pynenc.invocation.InvocationStatus, current_owner: str | None, attempted_owner: str | None, reason: str)

Bases: pynenc.exceptions.InvocationStatusError

Raised when attempting to modify an invocation without proper ownership.

This error occurs when:

• A non-owner tries to modify an owned invocation

• Attempting to acquire ownership without providing a runner ID

• A status requiring ownership has no owner set (invalid state)

Initialization

Create an invocation ownership error.

Parameters:

• from_status (InvocationStatus | None) – Current status

• to_status (InvocationStatus) – Attempted target status

• current_owner (str | None) – Current owner ID

• attempted_owner (str | None) – ID of runner attempting the change

• reason (str) – Explanation of why ownership was violated

__str__() → str

_to_json_dict() → dict[str, Any]

classmethod _from_json_dict(json_dict: dict[str, Any]) → pynenc.exceptions.InvocationStatusOwnershipError