crontab url caller

Dockers/cron-url-caller/entrypoint.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Call a URL on a cron schedule."""
+
+from __future__ import annotations
+
+import os
+import sys
+import time
+from datetime import datetime
+from typing import Any
+from urllib.error import URLError
+from urllib.request import Request, urlopen
+from zoneinfo import ZoneInfo
+
+from croniter import croniter
+
+CRON_MACROS = {
+    "@yearly": "0 0 1 1 *",
+    "@annually": "0 0 1 1 *",
+    "@monthly": "0 0 1 * *",
+    "@weekly": "0 0 * * 0",
+    "@daily": "0 0 * * *",
+    "@midnight": "0 0 * * *",
+    "@hourly": "0 * * * *",
+}
+
+
+def log(msg: str) -> None:
+    print(f"[{datetime.now().astimezone().isoformat(timespec='seconds')}] {msg}", file=sys.stderr)
+
+
+def zone() -> ZoneInfo:
+    name = os.environ.get("TZ") or "UTC"
+    try:
+        return ZoneInfo(name)
+    except Exception as exc:
+        raise SystemExit(f"Invalid TZ '{name}': {exc}") from exc
+
+
+def resolve_cron_expr() -> str:
+    cron = os.environ.get("CRON", "").strip()
+    if not cron:
+        raise SystemExit("CRON must be set (e.g. '0 3 * * *')")
+
+    expr = CRON_MACROS.get(cron.lower(), cron)
+    if not croniter.is_valid(expr):
+        raise SystemExit(f"Invalid CRON '{expr}' (expected a 5-field cron expression)")
+    return expr
+
+
+def resolve_url() -> str:
+    url = os.environ.get("URL", "").strip()
+    if not url:
+        raise SystemExit("URL must be set")
+    return url
+
+
+def timeout_seconds() -> int:
+    raw = os.environ.get("TIMEOUT", "30").strip()
+    try:
+        value = int(raw)
+    except ValueError as exc:
+        raise SystemExit(f"Invalid TIMEOUT '{raw}' (expected integer)") from exc
+    if value <= 0:
+        raise SystemExit("Invalid TIMEOUT (expected a positive integer)")
+    return value
+
+
+def run_on_startup() -> bool:
+    return os.environ.get("ON_STARTUP", "true").lower() in ("1", "true", "yes", "on")
+
+
+def next_run(expr: str, after: datetime) -> datetime:
+    return croniter(expr, after).get_next(datetime)
+
+
+def do_request(url: str, timeout: int) -> None:
+    req = Request(url, method="GET")
+    try:
+        with urlopen(req, timeout=timeout) as response:
+            log(f"GET {url} -> HTTP {response.status}")
+    except URLError as exc:
+        reason = exc.reason if hasattr(exc, "reason") else str(exc)
+        log(f"GET {url} failed: {reason}")
+
+
+def sleep_until(target: datetime, tz: ZoneInfo) -> None:
+    while True:
+        remaining = (target - datetime.now(tz)).total_seconds()
+        if remaining <= 0:
+            return
+        time.sleep(min(remaining, 60.0))
+
+
+def main() -> None:
+    url = resolve_url()
+    expr = resolve_cron_expr()
+    tz = zone()
+    timeout = timeout_seconds()
+    on_startup = run_on_startup()
+
+    log(f"Starting cron-url-caller (TZ={tz.key}, CRON='{expr}', URL='{url}', TIMEOUT={timeout}s)")
+
+    if on_startup:
+        do_request(url, timeout)
+
+    now = datetime.now(tz)
+    nxt = next_run(expr, now)
+    log(f"Next scheduled call at {nxt.isoformat(timespec='seconds')}")
+
+    while True:
+        sleep_until(nxt, tz)
+        do_request(url, timeout)
+        nxt = next_run(expr, datetime.now(tz))
+        log(f"Next scheduled call at {nxt.isoformat(timespec='seconds')}")
+
+
+if __name__ == "__main__":
+    main()