krz/hutch-stats

 1from __future__ import annotations
 2
 3from fastapi import Depends, Header, HTTPException, Request, status
 4from sqlalchemy.orm import Session
 5
 6from srht_contrib.config import Settings
 7from srht_contrib.db import get_db_session
 8from srht_contrib.jobs.poller import PollerService
 9from srht_contrib.utils.identity import ActorIdentityResolver
10
11
12def get_db(request: Request):
13    yield from get_db_session(request)
14
15
16def get_settings(request: Request) -> Settings:
17    settings = getattr(request.app.state, "settings", None)
18    if settings is None:
19        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Settings not configured.")
20    return settings
21
22
23def get_actor_identity_resolver(request: Request) -> ActorIdentityResolver:
24    resolver = getattr(request.app.state, "actor_identity_resolver", None)
25    if resolver is None:
26        raise HTTPException(
27            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
28            detail="Actor identity resolver not configured.",
29        )
30    return resolver
31
32
33def get_poller(request: Request) -> PollerService:
34    poller = getattr(request.app.state, "poller", None)
35    if poller is None:
36        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Poller not configured.")
37    return poller
38
39
40def require_api_key(
41    x_api_key: str | None = Header(default=None, alias="X-API-Key"),
42    settings: Settings = Depends(get_settings),
43) -> None:
44    if not settings.api_key or x_api_key != settings.api_key:
45        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key.")