spb/company-atlas

1"""Periodic task registry shared by the scheduler process.23The crawl scheduler (`catlas schedule`, services/scheduler.py) is the single long-running production process besides the API. Any4area (intelligence, alerts, metrics, retention, backups…) registers background work here instead of shipping its own daemon:56    from companyatlas.services.periodic import periodic78    @periodic("process-changes", every_s=30)9    async def process_changes_task(): ...1011    @periodic("metrics-hourly", cron="7 * * * *")12    async def metrics_task(): ...1314`every_s` tasks run on a fixed cadence (first run `initial_delay_s` after start); `cron` tasks use APScheduler cron syntax in15`settings.tz`. Tasks must be idempotent and bounded (do a batch, return). Exceptions are logged, never fatal. The scheduler16imports `companyatlas.services.registry_loader` (below) which imports every module that registers tasks — add yours there.17"""18from __future__ import annotations1920import asyncio21import importlib22import logging23import time24from collections.abc import Awaitable, Callable25from dataclasses import dataclass, field2627log = logging.getLogger(__name__)2829TaskFn = Callable[[], Awaitable[object]]303132@dataclass33class PeriodicTask:34    name: str35    fn: TaskFn36    every_s: float | None = None37    cron: str | None = None38    initial_delay_s: float = 5.039    exclusive: bool = True                  # never overlap with itself40    last_started_at: float | None = None41    last_finished_at: float | None = None42    last_error: str | None = None43    runs: int = 044    failures: int = 045    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)4647    async def run(self) -> None:48        if self.exclusive and self._lock.locked():49            return50        async with self._lock:51            self.last_started_at = time.time()52            try:53                await self.fn()54                self.runs += 155                self.last_error = None56            except Exception as exc:57                self.failures += 158                self.last_error = f"{exc.__class__.__name__}: {exc}"[:500]59                log.exception("periodic task failed", extra={"task": self.name})60            finally:61                self.last_finished_at = time.time()626364_TASKS: dict[str, PeriodicTask] = {}6566# Modules that register periodic tasks at import time. Areas append their module path here (one line each).67TASK_MODULES: list[str] = [68    "companyatlas.services.events",69    "companyatlas.services.metrics",70    "companyatlas.services.alerts",71    "companyatlas.services.retention",72    "companyatlas.services.llm.enrich",73    "companyatlas.services.signals",74    "companyatlas.services.trends",75    "companyatlas.services.repair",76    "companyatlas.services.enrichment",77    "companyatlas.commands.ops",78]798081def periodic(name: str, *, every_s: float | None = None, cron: str | None = None, initial_delay_s: float = 5.0, exclusive: bool = True):  # type: ignore[no-untyped-def]82    if not every_s and not cron:83        raise ValueError("periodic task needs every_s or cron")8485    def deco(fn: TaskFn) -> TaskFn:86        _TASKS[name] = PeriodicTask(name=name, fn=fn, every_s=every_s, cron=cron, initial_delay_s=initial_delay_s, exclusive=exclusive)87        return fn8889    return deco909192def load_task_modules() -> list[str]:93    """Import every registered task module (missing modules are skipped so areas can land independently)."""94    loaded: list[str] = []95    for mod in TASK_MODULES:96        try:97            importlib.import_module(mod)98            loaded.append(mod)99        except ModuleNotFoundError as exc:100            if exc.name and (mod == exc.name or mod.startswith(exc.name + ".")):101                continue102            log.exception("task module failed to import", extra={"task_module": mod})103        except Exception:104            log.exception("task module failed to import", extra={"task_module": mod})105    return loaded106107108def tasks() -> dict[str, PeriodicTask]:109    return _TASKS110111112def snapshot() -> list[dict[str, object]]:113    return [{"name": t.name, "every_s": t.every_s, "cron": t.cron, "runs": t.runs, "failures": t.failures, "last_started_at": t.last_started_at,114             "last_finished_at": t.last_finished_at, "last_error": t.last_error} for t in _TASKS.values()]115116117__all__ = ["TASK_MODULES", "PeriodicTask", "load_task_modules", "periodic", "snapshot", "tasks"]118