pynenc.trigger.conditions.cron

Context for time-based conditions like cron schedules.

This module provides time-based trigger conditions, including cron schedule triggers, that allow tasks to be executed at specific times or intervals.

Module Contents

Classes

CronContext

Context for time-based conditions like cron schedules.

CronCondition

Condition based on a cron schedule.

Data

DEFAULT_CHECK_WINDOW_SECONDS

DEFAULT_MIN_INTERVAL_SECONDS

DEFAULT_PRECISION_TOLERANCE_SECONDS

DEFAULT_STRICT_TIMING

API

pynenc.trigger.conditions.cron.DEFAULT_CHECK_WINDOW_SECONDS

60

pynenc.trigger.conditions.cron.DEFAULT_MIN_INTERVAL_SECONDS

50

pynenc.trigger.conditions.cron.DEFAULT_PRECISION_TOLERANCE_SECONDS

30

pynenc.trigger.conditions.cron.DEFAULT_STRICT_TIMING

False

class pynenc.trigger.conditions.cron.CronContext(*, timestamp: datetime.datetime | None = None, last_execution: datetime.datetime | None = None)[source]

Bases: pynenc.trigger.conditions.ConditionContext

Context for time-based conditions like cron schedules.

This class provides the context needed for evaluating time-based trigger conditions, including a check window in seconds and optional last execution time tracking.

Initialization

Create a time context with custom check window, timestamp, and optional last execution.

Parameters:

  • check_window_seconds – Duration in seconds to consider a match after scheduled time

  • timestamp – Specific timestamp to use (primarily for testing), defaults to current time

  • last_execution – Timestamp of the previous execution, or None if never executed

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

Create a serializable representation of this time 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.cron.CronContext[source]

Create a CronContext from parsed JSON data.

Parameters:

  • data – Dictionary with context data

  • app – Pynenc application instance

Returns:

A new CronContext instance

class pynenc.trigger.conditions.cron.CronCondition(cron_expression: str, check_window_seconds: int = DEFAULT_CHECK_WINDOW_SECONDS, min_interval_seconds: int = DEFAULT_MIN_INTERVAL_SECONDS, precision_tolerance_seconds: int = DEFAULT_PRECISION_TOLERANCE_SECONDS, strict_timing: bool = DEFAULT_STRICT_TIMING)[source]

Bases: pynenc.trigger.conditions.base.TriggerCondition[pynenc.trigger.conditions.cron.CronContext]

Condition based on a cron schedule.

Triggers a task at times matching a specified cron expression.

Initialization

Create a cron-based trigger condition.

Parameters:

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

  • check_window_seconds – Check window in seconds after scheduled time (default: 60)

  • min_interval_seconds – Minimum interval between executions (default: 50)

  • precision_tolerance_seconds – Precision tolerance for strict timing (default: 30)

  • strict_timing – Whether to enforce strict timing mode (default: False)

Raises:

ValueError – If the cron expression is invalid

context_type: ClassVar[type[pynenc.trigger.conditions.cron.CronContext]]

None

_validate_expression() → None[source]

Validate the cron expression.

property condition_id: str

Generate a unique ID for this cron condition.

Returns:

A string ID based on the cron expression

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

Create a serializable representation of this condition.

Parameters:

app – Pynenc application instance

Returns:

Dictionary with serialized condition data

classmethod _from_json(data: dict[str, Any], app: pynenc.app.Pynenc) → pynenc.trigger.conditions.cron.CronCondition[source]

Create a CronCondition from parsed JSON data.

Parameters:

  • data – Dictionary with condition data

  • app – Pynenc application instance

Returns:

A new CronCondition instance

Raises:

ValueError – If the data is invalid for this condition type

_is_satisfied_by(context: pynenc.trigger.conditions.cron.CronContext) → bool[source]

Check if the current time matches the cron schedule with flexible timing.

This method implements configurable timing behavior:

  • Uses check_window_seconds for the maximum time after scheduled execution

  • Respects min_interval_seconds to prevent too-frequent executions

  • Applies precision_tolerance_seconds for high-precision matching

  • Honors strict_timing mode when enabled

Parameters:

context – Time context with timestamp and optional last execution

Returns:

True if the timestamp satisfies the cron condition

_time_components_match(time1: datetime.datetime, time2: datetime.datetime) → bool[source]

Check if time components match based on cron expression fields.

This handles special cases when the exact timestamp matches the cron schedule.

Parameters:

  • time1 – First timestamp to compare

  • time2 – Second timestamp to compare

Returns:

True if relevant components match according to cron field count