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) -> CronA Cron from explicit field values (see Cron).
| Parameter | Type | Description |
|---|---|---|
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 |
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) -> boolWhether date (an aware datetime), read in the cron's zone, is one
of the cron's times.
| Parameter | Type | Description |
|---|---|---|
cron | Cron | |
date | datetime |
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) -> datetimeThe 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).
| Parameter | Type | Description |
|---|---|---|
cron | Cron | |
now | datetime | 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) -> datetimeThe last of the cron's times strictly before now (default: the
current time), in the cron's zone. Raises ValueError when no time
matches.
| Parameter | Type | Description |
|---|---|---|
cron | Cron | |
now | datetime | 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.
| Parameter | Type | Description |
|---|---|---|
cron | Cron | |
now | datetime | 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 currentformat#
format(cron, *, include_seconds=False) -> strThe 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.
| Parameter | Type | Description |
|---|---|---|
cron | Cron | |
include_seconds | bool = 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").
| Parameter | Type | Description |
|---|---|---|
expression | str | |
tz | tzinfo | 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) -> Cronparse, raising CronParseError for a bad expression.
| Parameter | Type | Description |
|---|---|---|
expression | str | |
tz | tzinfo | 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()