schedule.py

Overview

This module provides functionality to parse simplified cron-like schedule expressions and compute the next scheduled run time after a given reference datetime. It supports a minimal subset of cron syntax and enables lightweight task scheduling by converting human-readable cron expressions into structured data and performing an iterative search for the closest matching future datetime.

The core capabilities include:

• Parsing individual cron fields with support for wildcards (*), ranges (a-b), and comma-separated lists (a,b,c).

• Parsing complete five-field cron expressions representing minute, hour, day of month, month, and day of week.

• Calculating the next scheduled run datetime based on the parsed cron schedule, searching forward up to one year from a reference time.

This module integrates parsing and scheduling logic cohesively, serving as a foundation for time-based task execution within the system.

Classes and Functions

class CronField

A data container class representing a set of allowed integer values for a cron schedule field.

Attributes:

• values: List[int]
  A sorted list of unique integers that specify valid values for a particular cron time unit (e.g., minutes, hours).

Usage Example:

field = CronField(values=[0, 15, 30, 45])
print(field.values)  # Output: [0, 15, 30, 45]

_expand(field: str, *, min_v: int, max_v: int) -> CronField

Expands a single cron field string into a sorted set of integer values, supporting simplified cron syntax:

• * wildcard expands to all possible values between min_v and max_v.

• a-b range expands to all integers between a and b inclusive.

• a,b,c list expands to discrete integers listed.

Parameters:

• field — A string representing a cron field (e.g., "*", "1-5", "0,15,30").

• min_v — Minimum integer value allowed for the field.

• max_v — Maximum integer value allowed for the field.

Returns:

• CronField containing a sorted list of unique expanded integer values.

Example:

expand_result = _expand("1-3", min_v=0, max_v=5)
print(expand_result.values)  # Output: [1, 2, 3]

Implementation Notes:

• The function discerns the syntax type by checking for the presence of * or -.

• It converts all values to integers, removes duplicates, sorts them, and packages them into a CronField.

parse(cron: str) -> tuple[CronField, CronField, CronField, CronField, CronField]

Parses a complete five-field cron expression string and returns a tuple of expanded CronField objects representing each time unit.

Parameters:

• cron — A string containing five space-separated fields:
  minute hour day_of_month month day_of_week
  For example: "0 9 * * *" (meaning every day at 9:00 AM).

Returns:

• Tuple of five CronField instances corresponding to:
  (minute, hour, day_of_month, month, day_of_week)

Example:

minute, hour, dom, mon, dow = parse("0 9 * * *")
print(minute.values)  # Output: [0]
print(hour.values)    # Output: [9]
print(dom.values)     # Output: [1, 2, ..., 31] (all days of month)

Implementation Details:

• Splits the input string into the five fields.

• Uses _expand to parse each field with appropriate min and max values:

  • Minute: 0–59

  • Hour: 0–23

  • Day of month: 1–31

  • Month: 1–12

  • Day of week: 0–6 (where 0 corresponds to Monday as per Python's datetime.weekday() convention)

next_run(cron: str, *, since: datetime | None = None) -> datetime

Calculates the next datetime that matches the given cron schedule, searching forward starting just after the provided reference time.

Parameters:

• cron — Cron expression string with five fields (minute, hour, day of month, month, day of week).

• since (optional) — Reference datetime object from which to search for the next run time. Defaults to the current datetime if omitted.

Returns:

• A datetime object representing the earliest future time matching the schedule.

Raises:

• ValueError if no matching datetime is found within one year from since.

Example Usage:

from datetime import datetime

next_time = next_run("0 9 * * *", since=datetime(2024, 6, 1, 8, 59))
print(next_time)  # Output: 2024-06-01 09:00:00

Implementation Details:

• Calls parse() to get expanded cron fields.

• Normalizes the since time by zeroing seconds and microseconds, then adds one minute to start search.

• Iterates minute-by-minute up to a maximum of 366 * 24 * 60 minutes (~1 year).

• Checks if current datetime components (minute, hour, day, month, weekday) are contained in corresponding CronField values.

• Returns the first matching datetime.

• Raises an error if no match is found within the search limit.

Algorithmic Notes:

• The implementation uses a brute-force approach by incrementing minute-by-minute.

• It does not handle complex cron features like steps (*/5), names (e.g., Jan), or seconds fields.

• The search limit avoids infinite loops in cases of invalid or overly restrictive schedules.

Important Implementation Details and Algorithms

• Simplified Cron Syntax Support:
  The module supports only the simplest cron field syntaxes: wildcard (*), ranges (a-b), and comma-separated lists (a,b,c). This minimalism keeps parsing and matching logic straightforward and less error-prone.

• Use of dataclasses for CronField:
  The CronField class encapsulates the expanded values, providing type safety and clarity when passing around parsed schedule data.

• Iterative Search Bound:
  To avoid indefinite computation, the next_run function limits its search to one year of minute increments after the reference time.

• Python datetime.weekday() Convention:
  The day-of-week field is interpreted using Python's datetime.weekday() where Monday is 0 and Sunday is 6.

• No Seconds Field:
  The schedule fields do not include seconds; the search rounds up to the next minute.

• Sorted Unique Values:
  The _expand function returns values sorted and deduplicated to optimize membership checks during iteration.

Interaction with Other System Components

• The parsing and scheduling logic in this file is foundational for components that require task scheduling based on cron expressions. Other parts of the system can use parse() to interpret schedule strings and next_run() to determine when to trigger scheduled tasks.

• This module is closely related to the Cron Schedule Parsing, Cron Expression Expansion, and Next Run Calculation topics, providing their core logic within one cohesive file.

• Higher-level modules or demonstration scripts may invoke next_run to display or act upon the next scheduled execution time.

• This file operates independently from timing measurement utilities (Precise Task Timing) and current time retrieval helpers (Current Time Utility) but complements them by focusing on schedule computation.

Visual Diagram: Class and Function Structure

classDiagram
class CronField {
+values: List[int]
}
class _expand {
+field: str
+min_v: int
+max_v: int
+returns CronField
}
class parse {
+cron: str
+returns tuple[CronField, CronField, CronField, CronField, CronField]
}
class next_run {
+cron: str
+since: datetime | None
+returns datetime
}
parse --> CronField : returns
_expand --> CronField : returns
next_run --> parse : calls
next_run --> CronField : uses

This class diagram illustrates the relationships between the key entities and functions in the module:

• CronField acts as a data container.

• _expand parses individual fields into CronField.

• parse processes the full cron expression into a tuple of CronField.

• next_run depends on parse and uses the expanded fields to compute the next scheduled time.

References to Related Topics

• Cron Schedule Parsing — details the parsing of cron expressions into structured data.

• Cron Expression Expansion — explains the field expansion logic used within _expand.

• Next Run Calculation — describes the algorithm for computing the next run datetime.

• Next Run Computation — focuses on the iterative search algorithm implemented in next_run.

• Scheduling Demonstration — shows examples of using the parsing and next run functions in practice.