pynenc.trigger.trigger_builder

Builder classes for constructing trigger definitions.

This module provides a fluent interface for building trigger definitions, making it easier to define complex triggering conditions for tasks.

Module Contents

Classes

TriggerBuilder

Builder class for creating trigger definitions.

Functions

on_cron	Create a trigger builder for a cron schedule.
on_event	Create a trigger builder for an event.
on_status	Create a trigger builder for a task status change.

Data

C

API

pynenc.trigger.trigger_builder.C

‘TypeVar(…)’

class pynenc.trigger.trigger_builder.TriggerBuilder

Builder class for creating trigger definitions.

Provides a fluent interface for defining when and how tasks should be triggered, with methods for different trigger types and condition combinations.

Initialization

Initialize an empty trigger builder.

on_cron(cron_expression: str) → pynenc.trigger.trigger_builder.TriggerBuilder

Add a cron-based condition to trigger the task.

Parameters:

cron_expression – Standard cron expression (e.g., “0 0 * * *” for daily at midnight)

Returns:

This builder for method chaining

on_event(event_code: str, payload_filter: dict[str, Any] | collections.abc.Callable[[dict[str, Any]], bool] | None = None) → pynenc.trigger.trigger_builder.TriggerBuilder

Add an event-based condition to trigger the task.

Parameters:

• event_code – Type of event to listen for

• payload_filter – Optional payload arguments to match specific events or boolean callable

Returns:

This builder for method chaining

on_status(task: pynenc.task.Task, statuses: str | list[str] | pynenc.invocation.status.InvocationStatus | list[pynenc.invocation.status.InvocationStatus] = InvocationStatus.SUCCESS, call_arguments: dict[str, Any] | collections.abc.Callable[[dict[str, Any]], bool] | None = None) → pynenc.trigger.trigger_builder.TriggerBuilder

Add a task status condition to trigger after another task completes.

Parameters:

• task – The task to monitor

• statuses – Status(es) that trigger execution (default: InvocationStatus.SUCCESS)

• call_arguments – Optional arguments to match specific calls

Returns:

This builder for method chaining

on_any_result(task: pynenc.task.Task, call_arguments: dict[str, Any] | collections.abc.Callable[[dict[str, Any]], bool] | None = None) → pynenc.trigger.trigger_builder.TriggerBuilder

Add a result condition to trigger when a task finishes successfully.

Parameters:

• task – The task to monitor for results

• call_arguments – Optional filter for task arguments to match specific invocations

Returns:

This builder for method chaining

on_result(task: pynenc.task.Task, filter_result: Any | collections.abc.Callable[[Any], bool], call_arguments: dict[str, Any] | collections.abc.Callable[[dict[str, Any]], bool] | None = None) → pynenc.trigger.trigger_builder.TriggerBuilder

Add a result condition to trigger when a task returns a specific result or matches a filter.

This method allows defining conditions that trigger based on task results in several ways:

• Exact matching: Pass any value (including None) to match exactly

• Custom filtering: Pass a function that evaluates the result and returns a boolean

• Ignore results: Set ignore_result=True to accept any result value

Parameters:

• task – The task to monitor for results

• filter_result – Either a specific value to match or a callable that filters results

• call_arguments – Optional filter for task arguments to match specific invocations

Returns:

This builder for method chaining

on_exception(task: pynenc.task.Task, exception_types: list[str] | str | None = None, call_arguments: dict[str, Any] | collections.abc.Callable[[dict[str, Any]], bool] | None = None) → pynenc.trigger.trigger_builder.TriggerBuilder

Add an exception condition to trigger when a task fails with specific exceptions.

This method creates a condition that triggers when a monitored task fails with a specific exception type or any exception if none specified.

Parameters:

• task – The task to monitor for exceptions

• exception_types – Exception type name(s) to match, None to match any exception

• call_arguments – Optional filter for task arguments to match specific invocations

Returns:

This builder for method chaining

with_logic(logic: pynenc.trigger.conditions.CompositeLogic | str) → pynenc.trigger.trigger_builder.TriggerBuilder

Set the logic for combining multiple conditions.

Parameters:

logic – Logic to use (AND or OR)

Returns:

This builder for method chaining

with_arguments(args: dict[str, Any]) → pynenc.trigger.trigger_builder.TriggerBuilder

Set static arguments to pass to the task when triggered.

This is a legacy method - prefer with_args_static for new code.

Parameters:

args – Dictionary of static arguments to pass to the task

Returns:

This builder for method chaining

with_args_static(args: dict[str, Any]) → pynenc.trigger.trigger_builder.TriggerBuilder

Set static arguments to pass to the task when triggered.

Parameters:

args – Dictionary of static arguments to pass to the task

Returns:

This builder for method chaining

with_args_from_event(callback: collections.abc.Callable[[pynenc.trigger.conditions.EventContext], dict[str, Any]]) → pynenc.trigger.trigger_builder.TriggerBuilder

Define arguments based on the event context that triggered the task.

Parameters:

callback – Function that receives an EventContext and returns arguments

Returns:

This builder for method chaining

with_args_from_status(callback: collections.abc.Callable[[pynenc.trigger.conditions.StatusContext], dict[str, Any]]) → pynenc.trigger.trigger_builder.TriggerBuilder

Define arguments based on the task status context that triggered this task.

Parameters:

callback – Function that receives a StatusContext and returns arguments

Returns:

This builder for method chaining

with_args_from_result(callback: collections.abc.Callable[[pynenc.trigger.conditions.result.ResultContext], dict[str, Any]]) → pynenc.trigger.trigger_builder.TriggerBuilder

Define arguments based on the result context that triggered this task.

Parameters:

callback – Function that receives a ResultContext and returns arguments

Returns:

This builder for method chaining

with_args_from_exception(callback: collections.abc.Callable[[pynenc.trigger.conditions.exception.ExceptionContext], dict[str, Any]]) → pynenc.trigger.trigger_builder.TriggerBuilder

Define arguments based on the exception context that triggered this task.

The context type is inferred from the callback’s type annotation.

Parameters:

callback – Function that receives a specific ConditionContext and returns arguments

Returns:

This builder for method chaining

with_args_from_trigger_context(callback: collections.abc.Callable[[pynenc.trigger.trigger_context.TriggerContext], dict[str, Any]]) → pynenc.trigger.trigger_builder.TriggerBuilder

Define arguments with direct access to the trigger context.

Provides maximum flexibility by giving direct access to all valid conditions.

Parameters:

• callback – Function that receives the TriggerContext and returns arguments

• fallback – If True, ignores failures and continues with other providers

Returns:

This builder for method chaining

with_args_provider(provider: pynenc.trigger.arguments.argument_providers.ArgumentProvider) → pynenc.trigger.trigger_builder.TriggerBuilder

Add a custom argument provider.

Parameters:

provider – The argument provider to add

Returns:

This builder for method chaining

add_condition(condition: pynenc.trigger.conditions.base.TriggerCondition) → pynenc.trigger.trigger_builder.TriggerBuilder

Add a custom condition to the builder.

Parameters:

condition – The condition to add

Returns:

This builder for method chaining

build(task_id: pynenc.task.TaskId) → pynenc.trigger.trigger_definitions.TriggerDefinition

Build the trigger definition for a task.

Parameters:

task_id – ID of the task to trigger

Returns:

A trigger definition with the configured conditions

Raises:

ValueError – If no conditions are defined

pynenc.trigger.trigger_builder.on_cron(cron_expression: str) → pynenc.trigger.trigger_builder.TriggerBuilder

Create a trigger builder for a cron schedule.

Parameters:

cron_expression – Standard cron expression

Returns:

A trigger builder with the cron condition

pynenc.trigger.trigger_builder.on_event(event_code: str, required_params: dict[str, Any] | None = None) → pynenc.trigger.trigger_builder.TriggerBuilder

Create a trigger builder for an event.

Parameters:

• event_code – Type of event to listen for

• required_params – Optional parameters that must be present in the event payload

Returns:

A trigger builder with the event condition

pynenc.trigger.trigger_builder.on_status(task: pynenc.task.Task, statuses: str | list[str] | pynenc.invocation.status.InvocationStatus | list[pynenc.invocation.status.InvocationStatus] = InvocationStatus.SUCCESS, call_arguments: dict[str, Any] | None = None) → pynenc.trigger.trigger_builder.TriggerBuilder

Create a trigger builder for a task status change.

Parameters:

• task – The task to monitor

• statuses – Status(es) that trigger execution (default: InvocationStatus.SUCCESS)

• call_arguments – Optional arguments to match specific calls

Returns:

A trigger builder with the task status condition