API reference: Date/time triggers

class schedium.triggers.datetime.AtDateTime(run_date)[source][source]

Bases: BaseTrigger

One-shot trigger that fires at/after a specific datetime.

AtDateTime matches when now >= run_date.

Unlike cadence-based triggers (like Every), AtDateTime is intended for one-shot schedules. It is also safe if the scheduler starts late: the first evaluation after run_date will match, and deduplication ensures it runs only once.

Parameters:
run_datedatetime

The target datetime.

Parameters:

run_date (datetime)

Notes

Granularity.EXACT

AtDateTime declares EXACT as both its required and fallback granularity.

Deduplication token

When an AtDateTime is present anywhere in a trigger tree, schedium uses a token tied to run_date (rather than a time bucket). This ensures the job is treated as a true one-shot even if evaluated multiple times.

Composing with constraints

You can AND AtDateTime with constraints like On or BetweenDateTime to prevent late execution outside an intended window.

Examples

Run once at a specific time

>>> from datetime import datetime
>>> from schedium import AtDateTime
>>> trigger = AtDateTime(datetime(2026, 2, 8, 12, 0, 0))

Run once, but only if we’re still inside a window

>>> from datetime import datetime
>>> from schedium import AtDateTime, BetweenDateTime
>>> trigger = (
...     AtDateTime(datetime(2026, 2, 8, 12, 0, 0))
...     & BetweenDateTime(
...         start_date=datetime(2026, 2, 8, 0, 0, 0),
...         end_date=datetime(2026, 2, 8, 23, 59, 59),
...     )
... )
required_granularity()[source][source]
Return type:

Granularity

fallback_granularity()[source][source]
Return type:

Granularity

matches(now)[source][source]
Parameters:

now (datetime)

Return type:

bool

next_window(after, *, max_iterations=100000)[source][source]

Return the next validity window whose start is >= after.

For most constraint-style triggers, the default implementation:

  1. finds the next matching time using a forward scan at an inferred granularity, then

  2. returns a single-bucket window at that granularity.

Parameters:
afterdatetime

Lower bound (inclusive) for the returned window start.

max_iterationsint, default 100_000

Safety cap used by some triggers/combinators that scan forward.

Returns:
schedium.types.time_window.TimeWindow | None

Next validity window, or None if no future window exists.

Raises:
ValueError

If max_iterations <= 0.

schedium.exceptions.NextRunMaxIterationsReached

If a forward scan exceeds max_iterations.

Parameters:
  • after (datetime)

  • max_iterations (int)

Return type:

TimeWindow | None

class schedium.triggers.datetime.BetweenDateTime(start_date, end_date)[source][source]

Bases: BaseTrigger

Constraint trigger that matches within an inclusive datetime window.

BetweenDateTime is a constraint: it filters time but does not define a cadence by itself.

Parameters:
start_datedatetime

Inclusive lower bound of the window.

end_datedatetime

Inclusive upper bound of the window.

Parameters:
  • start_date (datetime)

  • end_date (datetime)

Notes

Inclusivity

The window is inclusive on both ends: start_date <= now <= end_date.

Validation

If start_date > end_date a ValueError is raised.

Granularity

This trigger reports a fallback granularity of SECOND. That fallback is only used for generic scanning logic in next_window(). BetweenDateTime also implements an efficient next_window() that returns:

  • start_date when queried before the window,

  • after when queried inside the window,

  • None when queried after the window.

Timezones

Keep start_date, end_date, and the scheduler’s now either all timezone-aware (same tz) or all naive; comparing mixed values raises in Python.

Examples

Allow a job to run only during a maintenance window

>>> from datetime import datetime, timezone
>>> from schedium import BetweenDateTime, Every
>>> window = BetweenDateTime(
...     start_date=datetime(2026, 2, 8, 1, 0, tzinfo=timezone.utc),
...     end_date=datetime(2026, 2, 8, 2, 0, tzinfo=timezone.utc),
... )
>>> trigger = Every(unit="minute", interval=1) & window
matches(now)[source][source]
Parameters:

now (datetime)

Return type:

bool

fallback_granularity()[source][source]
Return type:

Granularity

next_window(after, *, max_iterations=100000)[source][source]

Return the next validity window whose start is >= after.

For most constraint-style triggers, the default implementation:

  1. finds the next matching time using a forward scan at an inferred granularity, then

  2. returns a single-bucket window at that granularity.

Parameters:
afterdatetime

Lower bound (inclusive) for the returned window start.

max_iterationsint, default 100_000

Safety cap used by some triggers/combinators that scan forward.

Returns:
schedium.types.time_window.TimeWindow | None

Next validity window, or None if no future window exists.

Raises:
ValueError

If max_iterations <= 0.

schedium.exceptions.NextRunMaxIterationsReached

If a forward scan exceeds max_iterations.

Parameters:
  • after (datetime)

  • max_iterations (int)

Return type:

TimeWindow | None