quartz_cron

quartz_cron

Quartz-style (6/7-field) cron parsing and DST-correct next-fire computation.

Domo's real trigger expressions are Quartz cron, not standard 5-field Unix cron: second minute hour day-of-month month day-of-week [year], using ? as an "unconstrained, see the other of day-of-month/day-of-week" placeholder, */n steps, comma lists, and named weekdays/months (MON, JAN). See crew-dcs issue #1505.

This is a lift-and-harden of the stdlib-only (zoneinfo + calendar, no new dependency) parser originally written in projects/monit-schedule-timezones/monit_schedule_timezones.py (PR #1504), kept there as a deliberately standalone duplicate — see that file's own module docstring. This module is the maintained, unit-tested version other library code should import.

An unparseable expression or an unrecognized field value raises QuartzCronParseError rather than returning a falsy "not a cron" result — a silent wrong answer (issue #1505's "reads as no schedule") is the exact bug class this module exists to avoid.

QuartzCronSchedule dataclass

QuartzCronSchedule(
    seconds: set[int] | None,
    minutes: set[int] | None,
    hours: set[int] | None,
    days_of_month: set[int] | None,
    months: set[int] | None,
    days_of_week: set[int] | None,
    years: set[int] | None,
)

A parsed Quartz-style 7-field cron expression.

Each field is None (unconstrained — the expression used * or ?) or a set[int] of allowed values. Day-of-month and day-of-week are combined with AND, not Quartz's OR-when-both-restricted rule: every real schedule observed on domo-community sets exactly one of the two to ?, so AND and OR agree on every real input, and AND is simpler.

compute_next_fire_time

compute_next_fire_time(
    cron: QuartzCronSchedule,
    after_utc: datetime,
    zone: ZoneInfo,
    *,
    max_years_ahead: int = 8
) -> datetime | None

Find the next fire time strictly after after_utc, as a tz-aware UTC datetime.

Rolls candidate year/month/day/hour/minute/second forward field-by-field (the standard cron "next fire" algorithm — carry into the next larger field whenever a field's current value isn't in its allowed set) using zone's wall-clock calendar throughout, and converts only the final match to UTC via zoneinfo. This is what keeps a schedule DST-correct: the candidate is always built as "this wall-clock time in zone", so a daily 08:00 schedule stays 08:00 local both before and after a DST transition, converting to two different UTC instants (a fixed-offset shift would get one of the two wrong).

A wall-clock time that does not exist (skipped by a spring-forward transition) is resolved by zoneinfo per PEP 495 (the pre-transition offset, fold=0) rather than advanced to the next real instant — no observed Domo schedule uses a DST-observing schedule_timezone (they are all UTC), so this edge case has no real-data test coverage here.

Returns None if no match exists within max_years_ahead years (e.g. day-of-month 31 constrained to February) rather than looping forever — this is a legitimate "no fire time" answer, distinct from an unparseable expression, which raises instead.

Source code in src/crew_dcs/utils/quartz_cron.py

def compute_next_fire_time(  # noqa: C901
    cron: QuartzCronSchedule,
    after_utc: dt.datetime,
    zone: ZoneInfo,
    *,
    max_years_ahead: int = 8,
) -> dt.datetime | None:
    """Find the next fire time strictly after `after_utc`, as a tz-aware UTC datetime.

    Rolls candidate year/month/day/hour/minute/second forward field-by-field
    (the standard cron "next fire" algorithm — carry into the next larger
    field whenever a field's current value isn't in its allowed set) using
    `zone`'s wall-clock calendar throughout, and converts only the final
    match to UTC via `zoneinfo`. This is what keeps a schedule DST-correct:
    the candidate is always built as "this wall-clock time in `zone`", so a
    daily 08:00 schedule stays 08:00 local both before and after a DST
    transition, converting to two different UTC instants (a fixed-offset
    shift would get one of the two wrong).

    A wall-clock time that does not exist (skipped by a spring-forward
    transition) is resolved by `zoneinfo` per PEP 495 (the pre-transition
    offset, fold=0) rather than advanced to the next real instant — no
    observed Domo schedule uses a DST-observing `schedule_timezone` (they
    are all `UTC`), so this edge case has no real-data test coverage here.

    Returns `None` if no match exists within `max_years_ahead` years (e.g.
    day-of-month 31 constrained to February) rather than looping forever —
    this is a legitimate "no fire time" answer, distinct from an
    unparseable expression, which raises instead.
    """
    local = (after_utc.astimezone(zone) + dt.timedelta(seconds=1)).replace(
        microsecond=0
    )
    year, month, day = local.year, local.month, local.day
    hour, minute, second = local.hour, local.minute, local.second
    horizon_year = year + max_years_ahead

    for _ in range(500_000):  # absolute safety cap, independent of horizon_year
        if year > horizon_year:
            return None

        if cron.years is not None and year not in cron.years:
            candidates = [y for y in cron.years if y >= year]
            year = min(candidates) if candidates else horizon_year + 1
            month, day, hour, minute, second = 1, 1, 0, 0, 0
            continue

        if cron.months is not None and month not in cron.months:
            candidates = [m for m in cron.months if m >= month]
            if candidates:
                month = min(candidates)
            else:
                year += 1
                month = min(cron.months)
            day, hour, minute, second = 1, 0, 0, 0
            continue

        days_in_month = calendar.monthrange(year, month)[1]
        if day > days_in_month:
            month += 1
            if month > 12:
                month = 1
                year += 1
            day, hour, minute, second = 1, 0, 0, 0
            continue

        dom_ok = cron.days_of_month is None or day in cron.days_of_month
        py_weekday = dt.date(year, month, day).weekday()  # Mon=0 .. Sun=6
        quartz_dow = 1 if py_weekday == 6 else py_weekday + 2  # Sun=1 .. Sat=7
        dow_ok = cron.days_of_week is None or quartz_dow in cron.days_of_week
        if not (dom_ok and dow_ok):
            day += 1
            hour, minute, second = 0, 0, 0
            continue

        if cron.hours is not None and hour not in cron.hours:
            candidates = [h for h in cron.hours if h >= hour]
            if candidates:
                hour = min(candidates)
            else:
                day += 1
                hour = min(cron.hours)
            minute, second = 0, 0
            continue

        if cron.minutes is not None and minute not in cron.minutes:
            candidates = [m for m in cron.minutes if m >= minute]
            if candidates:
                minute = min(candidates)
            else:
                hour, minute = hour + 1, min(cron.minutes)
                if hour > 23:
                    hour, day = 0, day + 1
            second = 0
            continue

        if cron.seconds is not None and second not in cron.seconds:
            candidates = [s for s in cron.seconds if s >= second]
            if candidates:
                second = min(candidates)
            else:
                minute, second = minute + 1, min(cron.seconds)
                if minute > 59:
                    minute, hour = 0, hour + 1
                    if hour > 23:
                        hour, day = 0, day + 1
            continue

        return dt.datetime(
            year, month, day, hour, minute, second, tzinfo=zone
        ).astimezone(dt.UTC)

    return None

compute_next_run_time

compute_next_run_time(
    expression: str,
    timezone: str | None,
    *,
    after_utc: datetime | None = None,
    max_years_ahead: int = 8
) -> datetime | None

Parse expression as Quartz cron and compute its next UTC fire time.

timezone defaults to "UTC" when falsy. Raises QuartzCronParseError on an unparseable expression or an unrecognized IANA timezone — never returns None for either case. Returns None only when the expression parses cleanly but has no fire time within max_years_ahead years (e.g. day-of-month 31 constrained to February).

Source code in src/crew_dcs/utils/quartz_cron.py

def compute_next_run_time(
    expression: str,
    timezone: str | None,
    *,
    after_utc: dt.datetime | None = None,
    max_years_ahead: int = 8,
) -> dt.datetime | None:
    """Parse `expression` as Quartz cron and compute its next UTC fire time.

    `timezone` defaults to `"UTC"` when falsy. Raises `QuartzCronParseError`
    on an unparseable expression or an unrecognized IANA `timezone` — never
    returns `None` for either case. Returns `None` only when the expression
    parses cleanly but has no fire time within `max_years_ahead` years (e.g.
    day-of-month 31 constrained to February).
    """
    zone_name = timezone or "UTC"
    try:
        zone = ZoneInfo(zone_name)
    except (ZoneInfoNotFoundError, ValueError) as exc:
        raise QuartzCronParseError(
            expression, f"unknown schedule timezone {zone_name!r}"
        ) from exc

    cron = parse_quartz_cron(expression)
    return compute_next_fire_time(
        cron,
        after_utc or dt.datetime.now(dt.UTC),
        zone,
        max_years_ahead=max_years_ahead,
    )

parse_quartz_cron

parse_quartz_cron(expression: str) -> QuartzCronSchedule

Parse a Quartz-style cron expression: 6 or 7 whitespace-separated fields.

second minute hour dayOfMonth month dayOfWeek [year]. Raises QuartzCronParseError (never a bare exception type, never a falsy "not a cron" result) on a wrong field count, an out-of-range value, or an unsupported token (L/W/#).

Source code in src/crew_dcs/utils/quartz_cron.py

def parse_quartz_cron(expression: str) -> QuartzCronSchedule:
    """Parse a Quartz-style cron expression: 6 or 7 whitespace-separated fields.

    `second minute hour dayOfMonth month dayOfWeek [year]`. Raises
    `QuartzCronParseError` (never a bare exception type, never a falsy
    "not a cron" result) on a wrong field count, an out-of-range value, or
    an unsupported token (`L`/`W`/`#`).
    """
    fields = expression.split()
    if len(fields) not in (6, 7):
        raise QuartzCronParseError(
            expression,
            "expected 6 or 7 whitespace-separated Quartz cron fields "
            f"(second minute hour dayOfMonth month dayOfWeek [year]), got "
            f"{len(fields)}",
        )
    if len(fields) == 6:
        fields = [*fields, "*"]

    second, minute, hour, dom, month, dow, year = fields

    return QuartzCronSchedule(
        seconds=_parse_cron_field(
            second,
            expression=expression,
            field_name="second",
            min_v=0,
            max_v=59,
            names=None,
        ),
        minutes=_parse_cron_field(
            minute,
            expression=expression,
            field_name="minute",
            min_v=0,
            max_v=59,
            names=None,
        ),
        hours=_parse_cron_field(
            hour,
            expression=expression,
            field_name="hour",
            min_v=0,
            max_v=23,
            names=None,
        ),
        days_of_month=_parse_cron_field(
            dom,
            expression=expression,
            field_name="day-of-month",
            min_v=1,
            max_v=31,
            names=None,
        ),
        months=_parse_cron_field(
            month,
            expression=expression,
            field_name="month",
            min_v=1,
            max_v=12,
            names=_QUARTZ_MONTH_NAMES,
        ),
        days_of_week=_parse_cron_field(
            dow,
            expression=expression,
            field_name="day-of-week",
            min_v=1,
            max_v=7,
            names=_QUARTZ_DOW_NAMES,
        ),
        years=_parse_cron_field(
            year,
            expression=expression,
            field_name="year",
            min_v=1970,
            max_v=2199,
            names=None,
        ),
    )