from __future__ import annotations from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import Select, func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.db.models import ( AppState, ProbeResult, SourceImportRun, Tracker, TrackerSource, ) async def get_app_state(session: AsyncSession, key: str) -> str | None: row = await session.get(AppState, key) return row.value if row else None async def set_app_state(session: AsyncSession, key: str, value: str) -> None: row = await session.get(AppState, key) if row is None: session.add(AppState(key=key, value=value, updated_at=datetime.now(UTC))) else: row.value = value row.updated_at = datetime.now(UTC) await session.commit() async def count_trackers(session: AsyncSession) -> int: result = await session.scalar(select(func.count()).select_from(Tracker)) return int(result or 0) async def latest_successful_import_at(session: AsyncSession) -> datetime | None: """Finished-at of the most recent import that actually applied data. `not_modified` counts as success: the upstream feed was reachable and simply unchanged, so the inventory is not stale. """ return await session.scalar( select(func.max(SourceImportRun.finished_at)).where( SourceImportRun.status.in_(("ok", "not_modified")) ) ) async def overview_stats(session: AsyncSession) -> dict[str, Any]: total = await count_trackers(session) by_status: dict[str, int] = {} rows = await session.execute( select(Tracker.current_status, func.count()).group_by(Tracker.current_status) ) for status, count in rows.all(): by_status[status] = count last_import = await session.scalar( select(SourceImportRun) .where(SourceImportRun.status.in_(("ok", "not_modified", "failed"))) .order_by(SourceImportRun.finished_at.desc()) .limit(1) ) last_probe = await session.scalar( select(ProbeResult).order_by(ProbeResult.checked_at.desc()).limit(1) ) return { "total": total, "up": by_status.get("up", 0), "degraded": by_status.get("degraded", 0), "down": by_status.get("down", 0), "unknown": by_status.get("unknown", 0), "last_import_at": last_import.finished_at if last_import else None, "last_probe_at": last_probe.checked_at if last_probe else None, } def trackers_query( *, q: str | None = None, scheme: str | None = None, status: str | None = None, source: str | None = None, sort: str = "score", order: str = "desc", ) -> Select[tuple[Tracker]]: stmt = select(Tracker).options(selectinload(Tracker.sources)) if q: like = f"%{q.strip()}%" stmt = stmt.where(Tracker.canonical_url.ilike(like) | Tracker.hostname.ilike(like)) if scheme: stmt = stmt.where(Tracker.scheme == scheme.lower()) if status: stmt = stmt.where(Tracker.current_status == status.lower()) if source: stmt = stmt.join(TrackerSource).where(TrackerSource.source_name == source) sort_map = { "url": Tracker.canonical_url, "status": Tracker.current_status, "latency": Tracker.last_latency_ms, "checked": Tracker.last_checked_at, "hostname": Tracker.hostname, "protocol": Tracker.scheme, } col = sort_map.get(sort, Tracker.last_checked_at) if order.lower() == "asc": stmt = stmt.order_by(col.asc().nullslast(), Tracker.id.asc()) else: stmt = stmt.order_by(col.desc().nullslast(), Tracker.id.desc()) return stmt async def get_tracker(session: AsyncSession, tracker_id: int) -> Tracker | None: result = await session.execute( select(Tracker) .options( selectinload(Tracker.sources), selectinload(Tracker.dns_records), ) .where(Tracker.id == tracker_id) ) return result.scalar_one_or_none() async def due_trackers(session: AsyncSession, limit: int = 50) -> list[Tracker]: now = datetime.now(UTC) result = await session.execute( select(Tracker) .where( Tracker.current_status != "unsupported", (Tracker.next_check_at.is_(None)) | (Tracker.next_check_at <= now), ) .order_by(Tracker.next_check_at.asc().nullsfirst()) .limit(limit) ) return list(result.scalars().all()) async def prune_probe_results(session: AsyncSession, retention_days: int) -> int: cutoff = datetime.now(UTC) - timedelta(days=retention_days) rows = await session.execute(select(ProbeResult).where(ProbeResult.checked_at < cutoff)) old = list(rows.scalars().all()) for row in old: await session.delete(row) await session.commit() return len(old)