pynenc.trigger.conditions.status

Status-based trigger conditions for Pynenc.

This module provides conditions that trigger based on task and call status changes, allowing tasks to be triggered when other tasks reach specific states.

Module Contents

Classes

StatusContext	Context for task status conditions.
StatusCondition	Condition based on task status changes.

API

class pynenc.trigger.conditions.status.StatusContext

Bases: pynenc.trigger.conditions.base.ConditionContext

Context for task status conditions.

Contains the task ID, call ID, invocation ID, status, and call arguments to evaluate status-based conditions.

task_id: str

None

call_id: str

None

invocation_id: str

None

arguments: pynenc.arguments.Arguments

None

status: pynenc.invocation.status.InvocationStatus

None

property context_id: str

classmethod from_invocation(invocation: pynenc.invocation.dist_invocation.DistributedInvocation, status: Optional[pynenc.invocation.status.InvocationStatus] = None) → pynenc.trigger.conditions.status.StatusContext

Create a StatusContext from a DistInvocation.

Parameters:

• invocation – The invocation to extract context from

• status – Optional status to override the invocation’s status

Returns:

A StatusContext with task ID, call ID, invocation ID, status, and call arguments

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

Create a serializable representation of this status 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.status.StatusContext

Create a StatusContext from parsed JSON data.

Parameters:

• data – Dictionary with context data

• app – Pynenc application instance

Returns:

A new StatusContext instance

class pynenc.trigger.conditions.status.StatusCondition(task_id: str, statuses: list[pynenc.invocation.status.InvocationStatus], arguments_filter: pynenc.trigger.arguments.ArgumentFilter)

Bases: pynenc.trigger.conditions.base.TriggerCondition[pynenc.trigger.conditions.status.StatusContext]

Condition based on task status changes.

Triggers when a task reaches a specific status, with optional filtering by call arguments.

Initialization

Create a task status trigger condition.

Parameters:

• task_id – ID of the task to monitor

• statuses – Status(es) that satisfy this condition

• arguments_filter – Optional filter for task call arguments

context_type: ClassVar[type[pynenc.trigger.conditions.status.StatusContext]]

None

get_source_task_ids() → set[str]

property condition_id: str

Generate a unique ID for this status condition.

Returns:

A string ID based on task ID, statuses and call arguments

_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.status.StatusCondition

Create a StatusCondition from parsed JSON data.

Parameters:

• data – Dictionary with condition data

• app – Pynenc application instance for deserializing complex arguments

Returns:

A new StatusCondition instance

Raises:

ValueError – If the data is invalid for this condition type

_is_satisfied_by(context: pynenc.trigger.conditions.status.StatusContext) → bool

Check if a task status change satisfies this condition.

A status change satisfies the condition when:

1. The task ID matches the condition’s monitored task

2. The status is in the list of monitored statuses

3. The arguments filter matches the call arguments

Parameters:

context – Status context with task ID, status, and call arguments

Returns:

True if the condition is satisfied

affects_task(task_id: str) → 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