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
|
from __future__ import annotations
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from apscheduler.schedulers.background import BackgroundScheduler
from fastapi import FastAPI
from sqlalchemy import Engine
from sqlalchemy.orm import Session, sessionmaker
from srht_contrib.api.routes_contributions import router as contributions_router
from srht_contrib.api.routes_health import router as health_router
from srht_contrib.api.routes_repositories import router as repositories_router
from srht_contrib.config import Settings, get_settings
from srht_contrib.db import make_engine, validate_db
from srht_contrib.jobs.poller import PollerService
from srht_contrib.logging import configure_logging
from srht_contrib.services.git import GitIngestionService
from srht_contrib.services.srht_client import SourceHutGraphQLClient
from srht_contrib.services.todo import TodoIngestionService
from srht_contrib.utils.identity import ActorIdentityResolver
def build_poller(settings: Settings) -> PollerService:
todo_client = SourceHutGraphQLClient(settings.todo_srht_endpoint, settings.srht_token)
git_client = SourceHutGraphQLClient(settings.git_srht_endpoint, settings.srht_token)
todo_service = TodoIngestionService(todo_client, settings)
git_service = GitIngestionService(git_client, settings)
return PollerService(todo_service=todo_service, git_service=git_service)
def create_app(
settings: Settings | None = None,
*,
engine: Engine | None = None,
session_factory: sessionmaker[Session] | None = None,
poller: PollerService | None = None,
) -> FastAPI:
app_settings = settings or get_settings()
configure_logging()
app_engine = engine or make_engine(app_settings)
app_session_factory = session_factory or sessionmaker(
bind=app_engine,
autoflush=False,
autocommit=False,
expire_on_commit=False,
)
actor_identity_resolver = ActorIdentityResolver(app_settings.actor_aliases_json)
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
validate_db(app_engine)
app_poller = poller or build_poller(app_settings)
app.state.poller = app_poller
app.state.settings = app_settings
app.state.engine = app_engine
app.state.session_factory = app_session_factory
app.state.actor_identity_resolver = actor_identity_resolver
scheduler: BackgroundScheduler | None = None
if app_settings.enable_scheduler:
scheduler = BackgroundScheduler()
scheduler.add_job(
_scheduled_poll,
"interval",
seconds=app_settings.poll_interval_seconds,
args=[app],
id="srht-poller",
replace_existing=True,
)
scheduler.start()
app.state.scheduler = scheduler
try:
yield
finally:
if scheduler is not None:
scheduler.shutdown(wait=False)
app_poller.todo_service.client.close()
app_poller.git_service.client.close()
app = FastAPI(title=app_settings.app_name, lifespan=lifespan)
app.include_router(health_router)
app.include_router(contributions_router)
app.include_router(repositories_router)
return app
def _scheduled_poll(app: FastAPI) -> None:
settings: Settings = app.state.settings
poller: PollerService = app.state.poller
session_factory: sessionmaker[Session] = app.state.session_factory
db = session_factory()
try:
poller.poll_tracked_actors(db, settings.default_actor)
finally:
db.close()
app = create_app()
|
