pynenc.trigger.trigger_definitions

Trigger definition classes for the Pynenc system.

This module defines the classes used to specify when and how tasks should be triggered. Each trigger definition links a task to one or more conditions that determine when it executes.

Module Contents

Classes

TriggerDefinition

Defines when and how a task should be triggered.

Data

logger

API

pynenc.trigger.trigger_definitions.logger

‘getLogger(…)’

class pynenc.trigger.trigger_definitions.TriggerDefinition(task_id: pynenc.identifiers.task_id.TaskId, condition_ids: list[str], logic: pynenc.trigger.conditions.CompositeLogic = CompositeLogic.AND, argument_provider: pynenc.trigger.arguments.argument_providers.ArgumentProvider | None = None)

Defines when and how a task should be triggered.

A trigger definition combines a task with one or more conditions and specifies when the task should be executed.

Initialization

Create a new trigger definition.

Parameters:

• task_id – ID of the task to trigger

• condition_ids – IDs of conditions that determine when to trigger the task

• logic – Logic to apply when evaluating multiple conditions

• argument_provider – Provider for generating arguments dynamically

__eq__(value: object) → bool

_generate_trigger_id() → str

Generate a deterministic ID based on the trigger content.

This ensures the same trigger definitions across different runner instances have the same ID.

Returns:

A deterministic trigger ID as a hex string

generate_trigger_run_ids(trigger_context: pynenc.trigger.trigger_context.TriggerContext) → list[str]

Generate unique IDs for this trigger execution based on the triggering conditions.

For AND logic: Returns a single ID representing all conditions collectively For OR logic: Returns one ID per satisfied condition

These IDs are used to ensure each valid condition triggers a task exactly once across multiple workers.

Parameters:

trigger_context – Context with valid conditions

Returns:

List of unique trigger run IDs

should_trigger(trigger_context: pynenc.trigger.trigger_context.TriggerContext) → bool

Determine if the task should be triggered based on the context.

Parameters:

trigger_context – Context with valid conditions

Returns:

True if the task should be triggered, False otherwise

get_arguments(trigger_context: pynenc.trigger.trigger_context.TriggerContext) → dict[str, Any]

Get the arguments to pass to the task when triggered.

Uses the argument provider to generate arguments based on the trigger context.

Parameters:

trigger_context – Context that triggered this definition

Returns:

Arguments to pass to the task

to_dto(app: pynenc.app.Pynenc) → pynenc.models.trigger_definition_dto.TriggerDefinitionDTO

Serialize this trigger definition to a DTO.

Parameters:

app – Pynenc application instance

Returns:

DTO representation of this trigger definition

classmethod from_dto(dto: pynenc.models.trigger_definition_dto.TriggerDefinitionDTO, app: pynenc.app.Pynenc) → pynenc.trigger.trigger_definitions.TriggerDefinition

Create a trigger definition instance from a DTO.

Parameters:

• dto – DTO containing serialized trigger definition

• app – Pynenc application instance for deserializing conditions

Returns:

A new TriggerDefinition instance

Raises:

ValueError – If the JSON data is invalid