src/genro_asgi/tasks/schedule.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Schedule parsing and next-run computation — the three task kinds.16 17 - ``every``: ``"30s" | "15m" | "2h" | "1d"`` — next = after + interval.18 - ``cron``: classic 5-field string (min hour dom month dow) with ``* , - /``,19 parsed in-house (~60 lines; croniter stays out — zero dependencies).20 System cron semantics: dow accepts 0-7 (0 and 7 = Sunday); when BOTH dom21 and dow are restricted a date matches if EITHER matches; evaluation is in22 LOCAL time, like system cron.23 - ``at``: a list of ISO timestamps (naive = local time), each fires once;24 an exhausted list yields no next run (a DERIVED state — the record stays).25 26 All computed instants are POSIX epoch seconds (the store's ``*_ts`` fields).27 No replay: the next run is always computed forward from ``after`` (now),28 never backfilled for downtime.29 """30 31 from __future__ import annotations32 33 from datetime import datetime, timedelta34 from typing import Any35 36 __all__ = ["TaskCadence", "EverySpec", "CronSpec", "AtSpec"]37 38 EVERY_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400}39 CRON_SEARCH_DAYS = 366 * 4 # a valid spec matches well within 4 years40 41 42 class EverySpec:43 """A fixed interval; every run is the previous instant plus ``seconds``."""44 45 def __init__(self, spec: Any) -> None:46 """Parse ``"30s" | "15m" | "2h" | "1d"`` into ``seconds``.47 48 Raises:49 ValueError: On a malformed spec, an unknown unit or a zero interval.50 """51 text = str(spec).strip()52 number, unit = text[:-1], text[-1:]53 if not number.isdigit() or unit not in EVERY_UNITS:54 raise ValueError(f"invalid every spec: {spec!r} (want <number><s|m|h|d>)")55 self.seconds = int(number) * EVERY_UNITS[unit]56 if self.seconds <= 0:57 raise ValueError(f"invalid every spec: {spec!r} (zero interval)")58 59 def get_next_run(self, after_ts: float) -> float | None:60 """The instant one interval after ``after_ts`` (epoch seconds)."""61 return after_ts + self.seconds62 63 64 class AtSpec:65 """A list of one-shot instants; an exhausted list has no next run."""66 67 def __init__(self, spec: Any) -> None:68 """Parse a list of ISO timestamps into sorted ``instants`` (naive = local).69 70 Raises:71 ValueError: If ``spec`` is not a list or an entry is not ISO-parseable.72 """73 if not isinstance(spec, list):74 raise ValueError("invalid at spec: want a list of ISO timestamps")75 instants = []76 for entry in spec:77 try:78 instants.append(datetime.fromisoformat(str(entry)).timestamp())79 except ValueError as exc:80 raise ValueError(f"invalid at timestamp: {entry!r}") from exc81 self.instants = sorted(instants)82 83 def get_next_run(self, after_ts: float) -> float | None:84 """The first instant strictly after ``after_ts``, or None once past them all."""85 for instant in self.instants:86 if instant > after_ts:87 return instant88 return None89 90 91 class CronSpec:92 """A parsed 5-field cron string; ``get_next_run`` walks to the next match."""93 94 def __init__(self, spec: str) -> None:95 """Parse ``"min hour dom month dow"`` into value sets.96 97 Raises:98 ValueError: On a wrong field count or a malformed/out-of-range field.99 """100 self.spec = spec101 fields = str(spec).split()102 if len(fields) != 5:103 raise ValueError(f"invalid cron spec: {spec!r} (want 5 fields)")104 self.minutes = self._parse_field(fields[0], 0, 59)105 self.hours = self._parse_field(fields[1], 0, 23)106 self.days = self._parse_field(fields[2], 1, 31)107 self.months = self._parse_field(fields[3], 1, 12)108 # dow: 0-7 on the wire, 7 folds onto Sunday=0109 self.weekdays = {v % 7 for v in self._parse_field(fields[4], 0, 7)}110 self.dom_restricted = fields[2] != "*"111 self.dow_restricted = fields[4] != "*"112 113 def _parse_field(self, field: str, lo: int, hi: int) -> set[int]:114 """One cron field -> the set of matching values (``* , - /``)."""115 values: set[int] = set()116 for part in field.split(","):117 step = 1118 body = part119 if "/" in part:120 body, step_text = part.split("/", 1)121 if not step_text.isdigit() or int(step_text) < 1:122 raise ValueError(f"invalid cron step in {self.spec!r}: {part!r}")123 step = int(step_text)124 if body == "*":125 start, end = lo, hi126 elif "-" in body:127 a, _, b = body.partition("-")128 if not a.isdigit() or not b.isdigit():129 raise ValueError(f"invalid cron range in {self.spec!r}: {part!r}")130 start, end = int(a), int(b)131 elif body.isdigit():132 # a bare value; with a step it opens a range to the top (vixie)133 start = int(body)134 end = hi if "/" in part else start135 else:136 raise ValueError(f"invalid cron field in {self.spec!r}: {part!r}")137 if not (lo <= start <= hi and lo <= end <= hi and start <= end):138 raise ValueError(f"cron value out of range in {self.spec!r}: {part!r}")139 values.update(range(start, end + 1, step))140 return values141 142 def get_next_run(self, after_ts: float) -> float | None:143 """The first matching instant strictly after ``after_ts`` (epoch seconds).144 145 Raises:146 ValueError: If nothing matches within ~4 years (an impossible date,147 e.g. ``"0 0 31 2 *"``).148 """149 candidate = datetime.fromtimestamp(after_ts).replace(second=0, microsecond=0)150 candidate += timedelta(minutes=1)151 for _ in range(CRON_SEARCH_DAYS):152 if candidate.month in self.months and self._day_matches(candidate):153 matched = self._first_time_from(candidate)154 if matched is not None:155 return matched.timestamp()156 candidate = (candidate + timedelta(days=1)).replace(hour=0, minute=0)157 raise ValueError(f"cron spec {self.spec!r}: no occurrence within 4 years")158 159 def _day_matches(self, day: datetime) -> bool:160 """System cron date rule: dom OR dow when both are restricted."""161 dom_ok = day.day in self.days162 dow_ok = (day.weekday() + 1) % 7 in self.weekdays # python Mon=0 -> cron Sun=0163 if self.dom_restricted and self.dow_restricted:164 return dom_ok or dow_ok165 return dom_ok if self.dom_restricted else dow_ok if self.dow_restricted else True166 167 def _first_time_from(self, candidate: datetime) -> datetime | None:168 """First (hour, minute) match within candidate's day, from its time on."""169 for hour in sorted(self.hours):170 if hour < candidate.hour:171 continue172 for minute in sorted(self.minutes):173 if hour == candidate.hour and minute < candidate.minute:174 continue175 return candidate.replace(hour=hour, minute=minute)176 return None177 178 179 class TaskCadence:180 """The cadence of a task record: its ``kind`` picks the spec that computes."""181 182 spec_classes: dict[str, type[EverySpec | CronSpec | AtSpec]] = {183 "every": EverySpec,184 "cron": CronSpec,185 "at": AtSpec,186 }187 188 def __init__(self, kind: str, spec: Any) -> None:189 """Build the spec of ``kind`` — the parse happens here, once.190 191 Args:192 kind: ``"every" | "cron" | "at"``.193 spec: The kind's spec — interval string, cron string, or ISO list.194 195 Raises:196 ValueError: On an unknown kind or a malformed spec.197 """198 spec_class = self.spec_classes.get(kind)199 if spec_class is None:200 raise ValueError(f"unknown schedule kind: {kind!r}")201 self.kind = kind202 self.spec = spec_class(spec)203 204 def get_next_run(self, after_ts: float) -> float | None:205 """The next due instant, or None (an exhausted ``at`` list)."""206 return self.spec.get_next_run(after_ts)