GitHub

API reference for cron.

Cron schedules: parse a cron expression into a Cron, test whether a datetime matches it, and find the next or previous matching time::

every_weekday_9am = cron.parse_unsafe("0 9 * * MON-FRI", "Europe/Berlin")
cron.next(every_weekday_9am, now)

Expressions have five fields (minute hour day month weekday) or six, with seconds first. Times are aware datetime values; a cron without a zone runs in the zone of the datetime it is given. schedule.cron repeats an effect on a cron.

Classes

Functions

make

make(*, minutes, hours, days, months, weekdays, seconds=(0,), and_=False, tz=None) -> Cron

A Cron from explicit field values (see Cron).

ParameterTypeDescription
minutesIterable[int]
hoursIterable[int]
daysIterable[int]
monthsIterable[int]
weekdaysIterable[int]
secondsIterable[int] = (0,)
and_bool = False
tztzinfo | None = None

Returns Cron

def make(
    *,
    minutes: Iterable[int],
    hours: Iterable[int],
    days: Iterable[int],
    months: Iterable[int],
    weekdays: Iterable[int],
    seconds: Iterable[int] = (0,),
    and_: bool = False,
    tz: tzinfo | None = None,
) -> Cron:
    """A ``Cron`` from explicit field values (see ``Cron``)."""
    return Cron(
        minutes=minutes,
        hours=hours,
        days=days,
        months=months,
        weekdays=weekdays,
        seconds=seconds,
        and_=and_,
        tz=tz,
    )

match

match(cron, date) -> bool

Whether date (an aware datetime), read in the cron's zone, is one of the cron's times.

ParameterTypeDescription
cronCron
datedatetime

Returns bool

def match(cron: Cron, date: datetime) -> bool:
    """Whether ``date`` (an aware datetime), read in the cron's zone, is one
    of the cron's times."""
    zone_of(date, "date")
    wall = (date if cron.tz is None else date.astimezone(cron.tz)).replace(tzinfo=None)
    fields = (
        (cron.seconds, wall.second),
        (cron.minutes, wall.minute),
        (cron.hours, wall.hour),
        (cron.months, wall.month),
    )
    return all(not allowed or value in allowed for allowed, value in fields) and _day_matches(
        cron, wall
    )

next

next(cron, now=None) -> datetime

The first of the cron's times strictly after now (default: the current time), in the cron's zone. A wall time a DST change repeats runs once, at its first occurrence; one a DST change skips runs at the matching time after the gap. Raises ValueError when no time matches (such as February 30).

ParameterTypeDescription
cronCron
nowdatetime | None = None

Returns datetime

def next(cron: Cron, now: datetime | None = None) -> datetime:
    """The first of the cron's times strictly after ``now`` (default: the
    current time), in the cron's zone. A wall time a DST change repeats
    runs once, at its first occurrence; one a DST change skips runs at the
    matching time after the gap. Raises ``ValueError`` when no time
    matches (such as February 30)."""
    return _Walk(cron, _start(cron, now), reverse=False).run()

prev

prev(cron, now=None) -> datetime

The last of the cron's times strictly before now (default: the current time), in the cron's zone. Raises ValueError when no time matches.

ParameterTypeDescription
cronCron
nowdatetime | None = None

Returns datetime

def prev(cron: Cron, now: datetime | None = None) -> datetime:
    """The last of the cron's times strictly before ``now`` (default: the
    current time), in the cron's zone. Raises ``ValueError`` when no time
    matches."""
    return _Walk(cron, _start(cron, now), reverse=True).run()

sequence

sequence(cron, now=None) -> Iterator[datetime]

The cron's times after now (default: the current time), forever.

ParameterTypeDescription
cronCron
nowdatetime | None = None

Returns Iterator[datetime]

def sequence(cron: Cron, now: datetime | None = None) -> Iterator[datetime]:
    """The cron's times after ``now`` (default: the current time), forever."""
    current = now
    while True:
        current = next(cron, current)
        yield current

format

format(cron, *, include_seconds=False) -> str

The cron expression for cron, compacting runs into ranges and steps. The seconds field is left out when it is just 0, unless include_seconds. The zone is not part of the expression.

ParameterTypeDescription
cronCron
include_secondsbool = False

Returns str

def format(cron: Cron, *, include_seconds: bool = False) -> str:
    """The cron expression for ``cron``, compacting runs into ranges and
    steps. The seconds field is left out when it is just 0, unless
    ``include_seconds``. The zone is not part of the expression."""
    fields = (cron.seconds, cron.minutes, cron.hours, cron.days, cron.months, cron.weekdays)
    segments = [_format_segment(values) for values in fields]
    if not include_seconds and cron.seconds == frozenset({0}):
        segments = segments[1:]
    return " ".join(segments)

parse

parse(expression, tz=None) -> Result[Cron, CronParseError]

Parse a cron expression of five fields (minute hour day month weekday) or six (with seconds first). Fields take *, values, ranges a-b, steps /n and lists a,b; months and weekdays also take three-letter names. tz is a zone or a zone string ("Europe/Berlin", "+02:00").

ParameterTypeDescription
expressionstr
tztzinfo | str | None = None

Returns Result[Cron, CronParseError]

def parse(expression: str, tz: tzinfo | str | None = None) -> Result[Cron, CronParseError]:
    """Parse a cron expression of five fields (minute hour day month
    weekday) or six (with seconds first). Fields take ``*``, values, ranges
    ``a-b``, steps ``/n`` and lists ``a,b``; months and weekdays also take
    three-letter names. ``tz`` is a zone or a zone string
    (``"Europe/Berlin"``, ``"+02:00"``)."""
    segments = expression.split()
    if len(segments) not in (5, 6):
        return Err(CronParseError("Invalid number of segments in cron expression", expression))
    if len(segments) == 5:
        segments.insert(0, "0")
    zone = zone_from_string(tz) if isinstance(tz, str) else tz
    if isinstance(tz, str) and zone is None:
        return Err(CronParseError("Invalid time zone in cron expression", tz))
    parsed: list[tuple[set[int], bool]] = []
    for segment, spec in zip(
        segments, (_SECOND, _MINUTE, _HOUR, _DAY, _MONTH, _WEEKDAY), strict=True
    ):
        result = _parse_segment(segment, spec)
        if isinstance(result, CronParseError):
            return Err(result)
        parsed.append(result)
    seconds, minutes, hours, (days, days_wild), months, (weekdays, weekdays_wild) = parsed
    return Ok(
        Cron(
            tz=zone,
            seconds=seconds[0],
            minutes=minutes[0],
            hours=hours[0],
            days=days,
            months=months[0],
            weekdays=weekdays,
            and_=(days_wild or weekdays_wild) and bool(days) and bool(weekdays),
        )
    )

parse_unsafe

parse_unsafe(expression, tz=None) -> Cron

parse, raising CronParseError for a bad expression.

ParameterTypeDescription
expressionstr
tztzinfo | str | None = None

Returns Cron

def parse_unsafe(expression: str, tz: tzinfo | str | None = None) -> Cron:
    """``parse``, raising ``CronParseError`` for a bad expression."""
    return parse(expression, tz).get_or_raise()