krz/hutch-stats

  1from __future__ import annotations
  2
  3from collections.abc import AsyncIterator
  4from contextlib import asynccontextmanager
  5import logging
  6
  7from apscheduler.schedulers.background import BackgroundScheduler
  8from fastapi import FastAPI
  9from sqlalchemy import Engine
 10from sqlalchemy.orm import Session, sessionmaker
 11
 12from srht_contrib.api.routes_contributions import router as contributions_router
 13from srht_contrib.api.routes_health import router as health_router
 14from srht_contrib.api.routes_repositories import router as repositories_router
 15from srht_contrib.config import Settings, get_settings
 16from srht_contrib.db import make_engine, validate_db
 17from srht_contrib.jobs.poller import PollerService
 18from srht_contrib.logging import configure_logging
 19from srht_contrib.services.git import GitIngestionService
 20from srht_contrib.services.srht_client import SourceHutGraphQLClient
 21from srht_contrib.services.todo import TodoIngestionService
 22from srht_contrib.utils.identity import ActorIdentityResolver
 23
 24
 25logger = logging.getLogger(__name__)
 26
 27
 28def build_poller(settings: Settings) -> PollerService:
 29    todo_client = SourceHutGraphQLClient(settings.todo_srht_endpoint, settings.srht_token, request_delay=settings.srht_request_delay_seconds)
 30    git_client = SourceHutGraphQLClient(settings.git_srht_endpoint, settings.srht_token, request_delay=settings.srht_request_delay_seconds)
 31    todo_service = TodoIngestionService(todo_client, settings)
 32    git_service = GitIngestionService(git_client, settings)
 33    return PollerService(todo_service=todo_service, git_service=git_service, settings=settings)
 34
 35
 36def create_app(
 37    settings: Settings | None = None,
 38    *,
 39    engine: Engine | None = None,
 40    session_factory: sessionmaker[Session] | None = None,
 41    poller: PollerService | None = None,
 42) -> FastAPI:
 43    app_settings = settings or get_settings()
 44    configure_logging()
 45    app_engine = engine or make_engine(app_settings)
 46    app_session_factory = session_factory or sessionmaker(
 47        bind=app_engine,
 48        autoflush=False,
 49        autocommit=False,
 50        expire_on_commit=False,
 51    )
 52    actor_identity_resolver = ActorIdentityResolver(app_settings.actor_aliases_json)
 53
 54    @asynccontextmanager
 55    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 56        validate_db(app_engine)
 57        app_poller = poller or build_poller(app_settings)
 58        app.state.poller = app_poller
 59        app.state.settings = app_settings
 60        app.state.engine = app_engine
 61        app.state.session_factory = app_session_factory
 62        app.state.actor_identity_resolver = actor_identity_resolver
 63        scheduler: BackgroundScheduler | None = None
 64        if app_settings.enable_scheduler:
 65            scheduler = BackgroundScheduler()
 66            scheduler.add_job(
 67                _scheduled_poll,
 68                "interval",
 69                seconds=app_settings.poll_interval_seconds,
 70                args=[app],
 71                id="srht-poller",
 72                replace_existing=True,
 73            )
 74            scheduler.start()
 75            _run_startup_poll(app)
 76        app.state.scheduler = scheduler
 77        try:
 78            yield
 79        finally:
 80            if scheduler is not None:
 81                scheduler.shutdown(wait=False)
 82            app_poller.todo_service.client.close()
 83            app_poller.git_service.client.close()
 84
 85    app = FastAPI(title=app_settings.app_name, lifespan=lifespan)
 86
 87    app.include_router(health_router)
 88    app.include_router(contributions_router)
 89    app.include_router(repositories_router)
 90    return app
 91
 92
 93def _scheduled_poll(app: FastAPI) -> None:
 94    settings: Settings = app.state.settings
 95    poller: PollerService = app.state.poller
 96    session_factory: sessionmaker[Session] = app.state.session_factory
 97    db = session_factory()
 98    try:
 99        poller.poll_tracked_actors(db, settings.default_actor)
100    finally:
101        db.close()
102
103
104def _run_startup_poll(app: FastAPI) -> None:
105    try:
106        _scheduled_poll(app)
107    except Exception:
108        logger.exception("Initial scheduled poll failed during application startup")
109
110
111app = create_app()