app/scheduler.py (view raw)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
from __future__ import annotations
import asyncio
import logging
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from app.collectors import import_all_sources
from app.config import Settings, get_settings
from app.db import get_session_factory
from app.db.queries import prune_probe_results
from app.probes import probe_due_batch
logger = logging.getLogger(__name__)
_scheduler: AsyncIOScheduler | None = None
_locks = {
"import": asyncio.Lock(),
"probe": asyncio.Lock(),
"prune": asyncio.Lock(),
}
async def job_import_sources() -> None:
if _locks["import"].locked():
logger.warning("import job already running; skipping")
return
async with _locks["import"]:
settings = get_settings()
factory = get_session_factory()
async with factory() as session:
await import_all_sources(session, settings)
async def job_probe_due() -> None:
if _locks["probe"].locked():
logger.warning("probe job already running; skipping")
return
async with _locks["probe"]:
settings = get_settings()
factory = get_session_factory()
async with factory() as session:
count = await probe_due_batch(session, settings)
logger.info("probed %s trackers", count)
async def job_prune() -> None:
if _locks["prune"].locked():
return
async with _locks["prune"]:
settings = get_settings()
factory = get_session_factory()
async with factory() as session:
removed = await prune_probe_results(session, settings.raw_result_retention_days)
logger.info("pruned %s probe results", removed)
def start_scheduler(settings: Settings | None = None) -> AsyncIOScheduler:
global _scheduler
settings = settings or get_settings()
if _scheduler is not None:
return _scheduler
scheduler = AsyncIOScheduler(timezone="UTC")
scheduler.add_job(
job_import_sources,
"cron",
day_of_week=settings.import_day_of_week,
hour=settings.import_hour_utc,
minute=0,
id="import_sources",
max_instances=1,
coalesce=True,
)
scheduler.add_job(
job_probe_due,
"interval",
seconds=60,
id="probe_due",
max_instances=1,
coalesce=True,
)
scheduler.add_job(
job_prune,
"interval",
hours=24,
id="prune_results",
max_instances=1,
coalesce=True,
)
scheduler.start()
_scheduler = scheduler
logger.info("scheduler started (single process / single worker required)")
return scheduler
def stop_scheduler() -> None:
global _scheduler
if _scheduler is not None:
_scheduler.shutdown(wait=False)
_scheduler = None
|