pynenc.trigger.conditions.exception

Exception-based trigger conditions for task execution.

This module defines conditions based on the exception raised by a task, enabling triggers when a task fails with specific exception types.

Module Contents

Classes

ExceptionContext	Context for exception-based conditions.
ExceptionCondition	Condition based on the exception raised by a task.

API

class pynenc.trigger.conditions.exception.ExceptionContext

Bases: pynenc.trigger.conditions.status.StatusContext

Context for exception-based conditions.

This context holds the data required to evaluate conditions based on exceptions raised by a task. It extends StatusContext to add exception information.

exception_type: str

None

exception_message: str

None

property context_id: str

_to_json(app: pynenc.app.Pynenc) → dict[str, Any]

Create a serializable representation of this exception context.

Parameters:

app – Pynenc application instance

Returns:

Dictionary with serialized context data

classmethod _from_json(data: dict[str, Any], app: pynenc.app.Pynenc) → pynenc.trigger.conditions.exception.ExceptionContext

Create an ExceptionContext from parsed JSON data.

Parameters:

• data – Dictionary with context data

• app – Pynenc application instance

Returns:

A new ExceptionContext instance

class pynenc.trigger.conditions.exception.ExceptionCondition(task_id: pynenc.identifiers.task_id.TaskId, arguments_filter: pynenc.trigger.arguments.ArgumentFilter, exception_types: list[str])

Bases: pynenc.trigger.conditions.base.TriggerCondition[pynenc.trigger.conditions.exception.ExceptionContext]

Condition based on the exception raised by a task.

Triggers a task when another task raises a specific exception type. This condition combines task status filtering (requiring FAILED) with exception type filtering. It can filter based on:

1. The task call arguments (to match specific task invocations)

2. The exception type raised by the task

Initialization

Create an exception condition.

Parameters:

• task_id – ID of the task to monitor

• arguments_filter – Filter for task call arguments

• exception_types – List of exception type names to match

context_type: ClassVar[type[pynenc.trigger.conditions.exception.ExceptionContext]]

None

property task_id: pynenc.identifiers.task_id.TaskId

Get the task ID that this condition monitors.

Returns:

Task ID string

property arguments_filter: pynenc.trigger.arguments.ArgumentFilter

Get the arguments filter for this condition.

Returns:

The arguments filter

property statuses: list[pynenc.invocation.status.InvocationStatus]

Get the statuses that this condition triggers on.

For exception conditions, this is always [InvocationStatus.FAILED]

Returns:

List of statuses

property condition_id: str

Generate a unique ID for this exception condition.

Returns:

A string ID based on task ID, arguments, and exception types

get_source_task_ids() → set[pynenc.identifiers.task_id.TaskId]

_to_json(app: pynenc.app.Pynenc) → dict[str, Any]

Create a serializable representation of this condition.

Parameters:

app – Pynenc application instance for serializing complex arguments

Returns:

Dictionary with serialized condition data

classmethod _from_json(data: dict[str, Any], app: pynenc.app.Pynenc) → pynenc.trigger.conditions.exception.ExceptionCondition

Create an ExceptionCondition from parsed JSON data.

Parameters:

• data – Dictionary with condition data

• app – Pynenc application instance for deserializing complex arguments

Returns:

A new ExceptionCondition instance

Raises:

ValueError – If the data is invalid for this condition type

_is_satisfied_by(context: pynenc.trigger.conditions.exception.ExceptionContext) → bool

Check if a task exception satisfies this condition.

Parameters:

context – Exception context with exception data

Returns:

True if the task status condition is satisfied and exception matches

affects_task(task_id: pynenc.identifiers.task_id.TaskId) → bool

Check if this condition is affected by a specific task.

Parameters:

task_id – ID of the task to check

Returns:

True if this condition watches the specified task