Next Run Computation

Purpose

This subtopic addresses the core challenge of determining the next scheduled run time for a task based on a cron-like expression. Within the broader scope of Next Run Calculation, it focuses specifically on the algorithmic process that iteratively searches forward from a reference datetime to find the earliest datetime matching all components of a parsed schedule. This solves the problem of converting flexible, human-readable cron schedules into concrete upcoming run datetimes, enabling precise scheduling in testing and automation environments.

Functionality

The core workflow involves:

1. Parsing the cron expression:
   The input string containing five fields (minute, hour, day of month, month, day of week) is parsed into discrete integer sets representing valid values for each field. This is done by the parse() function, which internally uses the helper _expand() to support simple syntaxes like * (all values), ranges (a-b), and lists (a,b,c).

2. Iterative search for next run time:
   Starting from the next minute after the given reference datetime (or current datetime if none provided), the algorithm advances minute-by-minute, checking whether the current datetime matches all schedule fields. This includes matching minute, hour, day, month, and weekday values.

3. Return or error:
   If a datetime matching all fields is found within one year (366 days) of searching, it is returned as the next run time. Otherwise, the function raises an error indicating no suitable datetime was found within the limit.

This approach ensures a straightforward, deterministic method to find the next valid run time for any supported cron expression without relying on complex external libraries or cron daemons.

Key Code Snippet

The iterative search in next_run() exemplifies this:

t = since.replace(second=0, microsecond=0) + timedelta(minutes=1)
for _ in range(366 * 24 * 60):  # search limit: 1 year
    if (t.minute in minute.values and
        t.hour in hour.values and
        t.day in dom.values and
        t.month in mon.values and
        t.weekday() in dow.values):
        return t
    t += timedelta(minutes=1)
raise ValueError("No run time found within a year")

The parse() function and _expand() helper support the schedule field parsing, converting user-friendly strings into usable integer sets:

def _expand(field: str, *, min_v: int, max_v: int) -> CronField:
    if field == "*":
        values = list(range(min_v, max_v + 1))
    elif "-" in field:
        a, b = field.split("-", 1)
        values = list(range(int(a), int(b) + 1))
    else:
        values = [int(x) for x in field.split(",")]
    return CronField(sorted(set(values)))

Integration

This subtopic is a fundamental computational engine underlying the Next Run Calculation topic. While the parent topic describes the overall purpose and scope of determining next run datetimes, the algorithmic details and iterative minute-by-minute search reside here.

It leverages the parsing logic from Cron Schedule Parsing to convert cron strings into discrete integer sets but extends beyond simple parsing by performing the actual datetime iteration and matching.

Moreover, this precise computation complements the [Scheduling Demonstration] subtopic, which shows how to apply these functions practically. Together, they provide both the theoretical and practical layers of scheduling.

This subtopic’s behavior also aligns with the project’s goal of minimal dependencies and simplicity, implementing a clear and understandable search without relying on external cron libraries.

Diagram

flowchart TD
Start["Start from 'since' datetime"]
Adjust["Set seconds, microseconds to zero\nAdd 1 minute"]
Loop["For each minute up to 1 year"]
Check["Check all schedule fields match?"]
Yes["Yes: return current datetime"]
No["No: increment by 1 minute"]
Error["No match found\nraise error"]
Start --> Adjust --> Loop
Loop --> Check
Check -->|Yes| Yes
Check -->|No| No
No --> Loop
Loop -.->|exhausted| Error