feat: make Django configs drive backups and scheduling
Treat SQL-backed Django models as the source of truth for pobsync configuration, exporting runtime YAML only as a compatibility layer for the existing engine. Add a database-driven scheduler command, Docker scheduler services, schedule run-state fields, and tests for scheduler, config export, and retention behavior.
This commit is contained in:
88
src/pobsync_backend/scheduler.py
Normal file
88
src/pobsync_backend/scheduler.py
Normal file
@@ -0,0 +1,88 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CronSchedule:
|
||||
minute: str
|
||||
hour: str
|
||||
day_of_month: str
|
||||
month: str
|
||||
day_of_week: str
|
||||
|
||||
|
||||
def parse_cron_expr(expr: str) -> CronSchedule:
|
||||
parts = expr.strip().split()
|
||||
if len(parts) != 5:
|
||||
raise ValueError("cron expression must have exactly 5 fields")
|
||||
return CronSchedule(*parts)
|
||||
|
||||
|
||||
def due_key(moment: datetime) -> str:
|
||||
return moment.strftime("%Y%m%d%H%M")
|
||||
|
||||
|
||||
def is_due(expr: str, moment: datetime) -> bool:
|
||||
schedule = parse_cron_expr(expr)
|
||||
cron_dow = (moment.weekday() + 1) % 7
|
||||
return (
|
||||
_field_matches(schedule.minute, moment.minute, 0, 59)
|
||||
and _field_matches(schedule.hour, moment.hour, 0, 23)
|
||||
and _field_matches(schedule.day_of_month, moment.day, 1, 31)
|
||||
and _field_matches(schedule.month, moment.month, 1, 12)
|
||||
and _field_matches(schedule.day_of_week, cron_dow, 0, 7, sunday_alias=True)
|
||||
)
|
||||
|
||||
|
||||
def _field_matches(field: str, value: int, min_value: int, max_value: int, sunday_alias: bool = False) -> bool:
|
||||
for part in field.split(","):
|
||||
if _part_matches(part.strip(), value, min_value, max_value, sunday_alias=sunday_alias):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _part_matches(part: str, value: int, min_value: int, max_value: int, sunday_alias: bool) -> bool:
|
||||
if not part:
|
||||
return False
|
||||
|
||||
step = 1
|
||||
base = part
|
||||
if "/" in part:
|
||||
base, step_s = part.split("/", 1)
|
||||
if not step_s.isdigit():
|
||||
return False
|
||||
step = int(step_s)
|
||||
if step <= 0:
|
||||
return False
|
||||
|
||||
if base == "*":
|
||||
start = min_value
|
||||
end = max_value
|
||||
elif "-" in base:
|
||||
start_s, end_s = base.split("-", 1)
|
||||
if not start_s.isdigit() or not end_s.isdigit():
|
||||
return False
|
||||
start = int(start_s)
|
||||
end = int(end_s)
|
||||
elif base.isdigit():
|
||||
start = end = int(base)
|
||||
else:
|
||||
return False
|
||||
|
||||
values = _normalize_values(range(start, end + 1), sunday_alias=sunday_alias)
|
||||
normalized_value = 0 if sunday_alias and value == 7 else value
|
||||
if normalized_value not in values:
|
||||
return False
|
||||
return (normalized_value - min(values)) % step == 0
|
||||
|
||||
|
||||
def _normalize_values(values: range, sunday_alias: bool) -> set[int]:
|
||||
out: set[int] = set()
|
||||
for value in values:
|
||||
if sunday_alias and value == 7:
|
||||
out.add(0)
|
||||
else:
|
||||
out.add(value)
|
||||
return out
|
||||
Reference in New Issue
Block a user