diff --git a/fablednetmon/core/__init__.py b/fablednetmon/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/fablednetmon/core/migrations.py b/fablednetmon/core/migrations.py new file mode 100644 index 0000000..19033f8 --- /dev/null +++ b/fablednetmon/core/migrations.py @@ -0,0 +1,24 @@ +from __future__ import annotations +from pathlib import Path + + +def get_core_head() -> str: + """Return the current head revision ID of the core migration scripts. + + Called at migration authoring time (when running ``alembic revision``), + not at application runtime. The returned ID should be captured statically + in a plugin's ``depends_on`` field. + """ + from alembic.config import Config + from alembic.script import ScriptDirectory + + migrations_dir = Path(__file__).parent.parent / "migrations" + cfg = Config() + cfg.set_main_option("script_location", str(migrations_dir)) + script = ScriptDirectory.from_config(cfg) + heads = script.get_heads() + if len(heads) != 1: + raise RuntimeError( + f"Expected exactly 1 core migration head, got {len(heads)}: {heads}" + ) + return heads[0] diff --git a/fablednetmon/core/scheduler.py b/fablednetmon/core/scheduler.py new file mode 100644 index 0000000..101e016 --- /dev/null +++ b/fablednetmon/core/scheduler.py @@ -0,0 +1,42 @@ +from __future__ import annotations +import asyncio +import logging +from dataclasses import dataclass +from typing import Callable, Coroutine + +logger = logging.getLogger(__name__) + + +@dataclass +class ScheduledTask: + name: str + coro_factory: Callable[[], Coroutine] + interval_seconds: int + run_on_startup: bool = False + + +async def start_scheduler(tasks: list[ScheduledTask]) -> None: + """Run scheduled tasks in a loop. Call with asyncio.create_task().""" + last_run: dict[str, float] = {} + + for task in tasks: + if task.run_on_startup: + logger.info(f"Startup task: {task.name}") + asyncio.create_task(_run_task(task)) + last_run[task.name] = asyncio.get_event_loop().time() + + while True: + now = asyncio.get_event_loop().time() + for task in tasks: + last = last_run.get(task.name, 0) + if now - last >= task.interval_seconds: + asyncio.create_task(_run_task(task)) + last_run[task.name] = now + await asyncio.sleep(1) + + +async def _run_task(task: ScheduledTask) -> None: + try: + await task.coro_factory() + except Exception: + logger.exception(f"Scheduled task {task.name!r} raised an exception")