croniter parses cron expressions and iterates over matching datetimes. pip install croniter. Basic: from croniter import croniter; cron = croniter("0 9 * * 1-5", start_time); cron.get_next(datetime). Fields: second(optional) minute hour day_of_month month day_of_week. Next: cron.get_next() → timestamp; cron.get_next(datetime) → datetime. Prev: cron.get_prev(datetime). Match: croniter.match("0 9 * * *", dt) → bool. Valid: croniter.is_valid("*/5 * * * *"). Timezone: croniter("0 9 * * 1-5", tz=pytz.timezone("US/Eastern")). Hash: croniter("H H * * *", hash_id="myapp") — deterministic random offset. Multiple: croniter("0 8,12,18 * * *", ...). Ranges: */5 every 5; 1-5 range; 1,3,5 list. L: "0 0 L * *" — last day. W: "0 0 15W * *" — nearest weekday. Second-level: croniter("*/30 * * * * *", hash_id="", second_at_beginning=True). Cursor: cron.set_current(new_start). cron.get_current(). Iterator: for dt in cron: .... Next_n: list(itertools.islice(cron.all_next(datetime), 5)). Expand: croniter.expand("0 9 * * 1-5") → canonical. Claude Code generates croniter schedule helpers, next-run calculators, and cron-based task dispatchers.
CLAUDE.md for croniter
## croniter Stack
- Version: croniter >= 1.4 | pip install croniter
- Next: croniter("expr", start).get_next(datetime) → next datetime
- Prev: .get_prev(datetime) → previous datetime
- Match: croniter.match("expr", dt) → bool
- Valid: croniter.is_valid("expr") → bool
- TZ: croniter("expr", tz=pytz.timezone("America/New_York"))
- Hash: croniter("H H * * *", hash_id="service") — deterministic jittercroniter Schedule Pipeline
# app/schedule.py — croniter parsing, next/prev run, match, timezone, task dispatcher
from __future__ import annotations
import datetime
import itertools
from typing import Any, Callable, Iterator
from croniter import croniter
# ─────────────────────────────────────────────────────────────────────────────
# 1. Core helpers
# ─────────────────────────────────────────────────────────────────────────────
UTC = datetime.timezone.utc
def next_run(expr: str, after: datetime.datetime | None = None) -> datetime.datetime:
"""
Return the next run time for a cron expression.
Example:
next_run("0 9 * * 1-5") → next weekday at 09:00 UTC
"""
start = after or datetime.datetime.now(tz=UTC)
return croniter(expr, start).get_next(datetime.datetime)
def prev_run(expr: str, before: datetime.datetime | None = None) -> datetime.datetime:
"""Return the most recent (previous) run time for a cron expression."""
start = before or datetime.datetime.now(tz=UTC)
return croniter(expr, start).get_prev(datetime.datetime)
def matches(expr: str, dt: datetime.datetime) -> bool:
"""Return True if dt matches the cron expression."""
return croniter.match(expr, dt)
def is_valid(expr: str) -> bool:
"""Return True if the cron expression is syntactically valid."""
return croniter.is_valid(expr)
def next_n(expr: str, n: int, after: datetime.datetime | None = None) -> list[datetime.datetime]:
"""Return the next n run times after the given start time."""
start = after or datetime.datetime.now(tz=UTC)
cron = croniter(expr, start)
return list(itertools.islice(cron.all_next(datetime.datetime), n))
def schedule_iter(
expr: str,
start: datetime.datetime | None = None,
direction: str = "next",
) -> Iterator[datetime.datetime]:
"""
Yield datetimes matching a cron expression indefinitely.
direction: "next" (forward) or "prev" (backward).
"""
current = start or datetime.datetime.now(tz=UTC)
cron = croniter(expr, current)
getter = cron.all_next if direction == "next" else cron.all_prev
yield from getter(datetime.datetime)
# ─────────────────────────────────────────────────────────────────────────────
# 2. Schedule inspection
# ─────────────────────────────────────────────────────────────────────────────
def time_until_next(expr: str, now: datetime.datetime | None = None) -> datetime.timedelta:
"""Return timedelta until the next run."""
t = now or datetime.datetime.now(tz=UTC)
return next_run(expr, after=t) - t
def describe_schedule(expr: str, n: int = 5) -> dict[str, Any]:
"""
Return a human-readable summary of a cron schedule.
"""
if not is_valid(expr):
return {"valid": False, "expr": expr}
now = datetime.datetime.now(tz=UTC)
upcoming = next_n(expr, n, after=now)
avg_interval = None
if len(upcoming) >= 2:
gaps = [(upcoming[i+1] - upcoming[i]).total_seconds()
for i in range(len(upcoming)-1)]
avg_interval = sum(gaps) / len(gaps)
return {
"valid": True,
"expr": expr,
"next": upcoming[0] if upcoming else None,
"upcoming": upcoming,
"avg_interval_seconds": avg_interval,
"time_until_next": time_until_next(expr, now),
}
def runs_within(
expr: str,
window_start: datetime.datetime,
window_end: datetime.datetime,
) -> list[datetime.datetime]:
"""
Return all datetimes where the schedule fires within [window_start, window_end].
"""
results = []
cron = croniter(expr, window_start)
while True:
dt = cron.get_next(datetime.datetime)
if dt > window_end:
break
results.append(dt)
return results
def daily_run_count(expr: str, date: datetime.date) -> int:
"""Count how many times a cron fires on a given date."""
start = datetime.datetime.combine(date, datetime.time.min, tzinfo=UTC)
end = datetime.datetime.combine(date, datetime.time.max, tzinfo=UTC)
return len(runs_within(expr, start, end))
# ─────────────────────────────────────────────────────────────────────────────
# 3. Timezone-aware scheduling
# ─────────────────────────────────────────────────────────────────────────────
def next_run_tz(
expr: str,
timezone_name: str,
after: datetime.datetime | None = None,
) -> datetime.datetime:
"""
Return the next run time in a specific timezone.
Requires: pip install pytz
Example:
next_run_tz("0 9 * * 1-5", "America/New_York")
"""
import pytz
tz = pytz.timezone(timezone_name)
start = after or datetime.datetime.now(tz=tz)
if start.tzinfo is None:
start = tz.localize(start)
return croniter(expr, start, hash_id=None).get_next(datetime.datetime)
# ─────────────────────────────────────────────────────────────────────────────
# 4. Task registry / dispatcher
# ─────────────────────────────────────────────────────────────────────────────
class ScheduledTask:
"""A task with a cron expression and a callable."""
def __init__(self, name: str, expr: str, fn: Callable, enabled: bool = True):
if not is_valid(expr):
raise ValueError(f"Invalid cron expression: {expr!r}")
self.name = name
self.expr = expr
self.fn = fn
self.enabled = enabled
def is_due(self, now: datetime.datetime | None = None) -> bool:
"""Return True if this task should run at the given time (truncated to minute)."""
t = now or datetime.datetime.now(tz=UTC)
t_trunc = t.replace(second=0, microsecond=0)
return self.enabled and matches(self.expr, t_trunc)
def next_run(self, after: datetime.datetime | None = None) -> datetime.datetime:
return next_run(self.expr, after=after)
def run(self, **kwargs) -> Any:
return self.fn(**kwargs)
class TaskRegistry:
"""
Registry of ScheduledTask objects.
Call dispatch() once per minute to run due tasks.
Usage:
registry = TaskRegistry()
@registry.task("*/5 * * * *")
def poll_api():
...
while True:
registry.dispatch()
time.sleep(60)
"""
def __init__(self):
self._tasks: dict[str, ScheduledTask] = {}
def register(
self,
name: str,
expr: str,
fn: Callable,
enabled: bool = True,
) -> ScheduledTask:
task = ScheduledTask(name, expr, fn, enabled=enabled)
self._tasks[name] = task
return task
def task(self, expr: str, name: str | None = None, enabled: bool = True):
"""Decorator factory for registering tasks."""
def decorator(fn: Callable) -> Callable:
task_name = name or fn.__name__
self.register(task_name, expr, fn, enabled=enabled)
return fn
return decorator
def dispatch(self, now: datetime.datetime | None = None) -> list[str]:
"""Run all due tasks. Returns list of task names that ran."""
t = now or datetime.datetime.now(tz=UTC)
ran = []
for task in self._tasks.values():
if task.is_due(t):
try:
task.run()
ran.append(task.name)
except Exception as exc:
print(f"[TaskRegistry] {task.name} failed: {exc}")
return ran
def next_runs(self, n: int = 3) -> dict[str, list[datetime.datetime]]:
"""Return upcoming run times for every registered task."""
return {name: next_n(task.expr, n) for name, task in self._tasks.items()}
def due_tasks(self, now: datetime.datetime | None = None) -> list[str]:
"""Return names of tasks due now."""
t = now or datetime.datetime.now(tz=UTC)
return [name for name, task in self._tasks.items() if task.is_due(t)]
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
now = datetime.datetime(2028, 4, 17, 8, 0, 0, tzinfo=UTC)
print("=== next_run ===")
# Every weekday at 09:00
n = next_run("0 9 * * 1-5", after=now)
print(f" Next weekday 09:00: {n}")
# Every 15 minutes
n2 = next_run("*/15 * * * *", after=now)
print(f" Next */15: {n2}")
print("\n=== next_n ===")
upcoming = next_n("0 9 * * 1-5", 5, after=now)
for dt in upcoming:
print(f" {dt.strftime('%a %Y-%m-%d %H:%M')}")
print("\n=== matches ===")
mon_9am = datetime.datetime(2028, 4, 17, 9, 0, tzinfo=UTC)
print(f" Mon 09:00 matches '0 9 * * 1-5': {matches('0 9 * * 1-5', mon_9am)}")
print(f" Mon 10:00 matches '0 9 * * 1-5': {matches('0 9 * * 1-5', mon_9am.replace(hour=10))}")
print("\n=== time_until_next ===")
delta = time_until_next("0 9 * * *", now=now)
print(f" Until next daily 09:00: {delta}")
print("\n=== runs_within (today) ===")
today_start = datetime.datetime(2028, 4, 17, 0, 0, tzinfo=UTC)
today_end = datetime.datetime(2028, 4, 17, 23, 59, tzinfo=UTC)
runs = runs_within("0 */4 * * *", today_start, today_end)
print(f" Every 4h runs today ({len(runs)}): {[r.strftime('%H:%M') for r in runs]}")
print("\n=== describe_schedule ===")
info = describe_schedule("30 8 * * 1-5", n=3)
print(f" expr={info['expr']}")
print(f" next={info['next']}")
print(f" avg_interval={info['avg_interval_seconds']} s")
print("\n=== TaskRegistry ===")
registry = TaskRegistry()
log: list[str] = []
@registry.task("*/5 * * * *", name="poll_api")
def poll_api():
log.append("poll_api")
@registry.task("0 * * * *", name="hourly_report")
def hourly_report():
log.append("hourly_report")
@registry.task("0 0 * * *", name="daily_cleanup")
def daily_cleanup():
log.append("daily_cleanup")
# Simulate 08:00 — nothing due (hours not */5 boundary)
t1 = datetime.datetime(2028, 4, 17, 8, 0, tzinfo=UTC)
ran = registry.dispatch(now=t1)
print(f" 08:00 ran: {ran}")
# Simulate 08:05 — poll due
t2 = datetime.datetime(2028, 4, 17, 8, 5, tzinfo=UTC)
ran2 = registry.dispatch(now=t2)
print(f" 08:05 ran: {ran2}")
# Simulate 09:00 — poll + hourly
t3 = datetime.datetime(2028, 4, 17, 9, 0, tzinfo=UTC)
ran3 = registry.dispatch(now=t3)
print(f" 09:00 ran: {ran3}")
print("\n=== Next runs per task ===")
for task_name, times in registry.next_runs(3).items():
print(f" {task_name}: {[t.strftime('%Y-%m-%d %H:%M') for t in times]}")For the APScheduler alternative — APScheduler is a full scheduler with background threads, async support, persistent job stores, and a cron trigger; croniter is the underlying cron iterator that APScheduler itself uses — use croniter when you only need to compute next/prev run times, check if a time matches a schedule, or iterate over future run times without a running scheduler process. For the schedule library alternative — the schedule library provides a simple human-readable API (schedule.every().monday.at("09:00").do(fn)) but has limited cron expression support; croniter supports the full 5-field (or 6-field with seconds) cron syntax including L, W, and # modifiers for last-day, nearest-weekday, and nth-weekday expressions. The Claude Skills 360 bundle includes croniter skill sets covering next_run()/prev_run()/matches()/is_valid(), next_n() and schedule_iter() generators, time_until_next()/describe_schedule(), runs_within()/daily_run_count() window queries, next_run_tz() timezone-aware scheduling, ScheduledTask is_due()/next_run()/run(), and TaskRegistry with @task decorator, dispatch(), and next_runs(). Start with the free tier to try cron schedule code generation.