schedule

schedule

Planning and scheduling utilities for Snowflake stored procedures and DAGs.

Groups

Calendar French public-holiday calendar, business-day helpers (no external deps). Cron Human-readable cron formatting, next-schedule computation.

Optional dep for cron execution: croniter (pip install croniter).

business_days_between(start, end, country='FR')

Count business days in [start, end) -- start inclusive, end exclusive.

Source code in src/pinky_core/schedule.py

def business_days_between(start: date, end: date, country: str = "FR") -> int:
    """Count business days in [start, end) -- start inclusive, end exclusive."""
    if start >= end:
        return 0
    return sum(
        1
        for i in range((end - start).days)
        if is_business_day(start + timedelta(days=i), country=country)
    )

cron_to_human(expr)

Return a human-readable description of a standard 5-field cron expression.

Format: minute hour dom month dow (Snowflake / POSIX convention).

Examples

"0 9 * * 1-5" -> "weekdays (Mon-Fri), at 09:00" "30 6 * * " -> "every day, at 06:30" "0 0 1 * " -> "on day 1 of every month, at 00:00"

Source code in src/pinky_core/schedule.py

def cron_to_human(expr: str) -> str:
    """Return a human-readable description of a standard 5-field cron expression.

    Format: minute hour dom month dow  (Snowflake / POSIX convention).

    Examples
    --------
    "0 9 * * 1-5"  -> "weekdays (Mon-Fri), at 09:00"
    "30 6 * * *"   -> "every day, at 06:30"
    "0 0 1 * *"    -> "on day 1 of every month, at 00:00"
    """
    parts = expr.strip().split()
    if len(parts) != 5:
        raise ValueError(f"Expected 5-field cron, got {len(parts)} fields: {expr!r}")

    minute, hour, dom, month, dow = parts

    # ── time ──
    if minute == "*" and hour == "*":
        time_str = "every minute"
    elif minute.isdigit() and hour.isdigit():
        time_str = f"at {int(hour):02d}:{int(minute):02d}"
    elif hour.isdigit():
        time_str = f"at hour {int(hour):02d}, minute {minute}"
    else:
        time_str = f"at minute {minute}"

    # ── frequency ──
    if dom == "*" and month == "*" and dow == "*":
        freq_str = "every day"
    elif dow != "*" and dom == "*":
        freq_str = _dow_to_human(dow)
    elif dom != "*":
        freq_str = _dom_to_human(dom, month)
    else:
        freq_str = f"on days {dom} of month {month}, weekdays {dow}"

    return f"{freq_str}, {time_str}"

is_business_day(d, country='FR')

Return True if d is a weekday and not a public holiday.

Source code in src/pinky_core/schedule.py

def is_business_day(d: date, country: str = "FR") -> bool:
    """Return True if d is a weekday and not a public holiday."""
    if d.weekday() >= 5:
        return False
    if country == "FR":
        return d not in public_holidays_fr(d.year)
    raise ValueError(f"Unsupported country: {country!r}")

is_in_schedule(expr, dt=None)

Return True if dt (defaults to now UTC, minute-truncated) matches the cron expression.

Requires: croniter -- pip install croniter

Source code in src/pinky_core/schedule.py

def is_in_schedule(expr: str, dt: datetime | None = None) -> bool:
    """Return True if dt (defaults to now UTC, minute-truncated) matches the cron expression.

    Requires: croniter -- pip install croniter
    """
    try:
        from croniter import croniter
    except ImportError as exc:
        raise ImportError(
            "is_in_schedule requires 'croniter': pip install croniter"
        ) from exc
    check = (dt or datetime.utcnow()).replace(second=0, microsecond=0)
    prev: datetime = croniter(expr, check - timedelta(minutes=1)).get_next(datetime)
    return prev == check

next_business_day(d, n=1, country='FR')

Return the n-th business day strictly after d.

Source code in src/pinky_core/schedule.py

def next_business_day(d: date, n: int = 1, country: str = "FR") -> date:
    """Return the n-th business day strictly after d."""
    current, count = d, 0
    while count < n:
        current += timedelta(days=1)
        if is_business_day(current, country=country):
            count += 1
    return current

next_schedule(expr, from_dt=None)

Return the next scheduled datetime after from_dt (defaults to now UTC).

Requires: croniter -- pip install croniter

Source code in src/pinky_core/schedule.py

def next_schedule(expr: str, from_dt: datetime | None = None) -> datetime:
    """Return the next scheduled datetime after from_dt (defaults to now UTC).

    Requires: croniter -- pip install croniter
    """
    try:
        from croniter import croniter
    except ImportError as exc:
        raise ImportError(
            "next_schedule requires 'croniter': pip install croniter"
        ) from exc
    base = from_dt or datetime.utcnow()
    result: datetime = croniter(expr, base).get_next(datetime)
    return result

public_holidays_fr(year)

Return the set of French public holidays (metropole) for the given year.

Source code in src/pinky_core/schedule.py

def public_holidays_fr(year: int) -> frozenset[date]:
    """Return the set of French public holidays (metropole) for the given year."""
    easter = _easter(year)
    return frozenset(
        {
            # Fixed
            date(year, 1, 1),  # Jour de l'An
            date(year, 5, 1),  # Fete du Travail
            date(year, 5, 8),  # Victoire 1945
            date(year, 7, 14),  # Fete Nationale
            date(year, 8, 15),  # Assomption
            date(year, 11, 1),  # Toussaint
            date(year, 11, 11),  # Armistice
            date(year, 12, 25),  # Noel
            # Easter-based
            easter + timedelta(days=1),  # Lundi de Paques
            easter + timedelta(days=39),  # Ascension
            easter + timedelta(days=50),  # Lundi de Pentecote
        }
    )