100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import time as time_module
|
|
import uuid
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .config import settings
|
|
from .dependencies import Db
|
|
from .logging_config import configure_logging
|
|
from .readiness import readiness_report
|
|
from .routers.activity import router as activity_router
|
|
from .routers.admin import router as admin_router
|
|
from .routers.catalog import router as catalog_router
|
|
from .routers.media import router as media_router
|
|
from .routers.public_data import router as public_data_router
|
|
from .routers.submissions import router as submissions_router
|
|
from .storage import client as storage_client
|
|
from .submission_security import check_rate_limit
|
|
from .submission_security import is_trusted_proxy as _is_trusted_proxy
|
|
|
|
|
|
configure_logging(settings.log_level)
|
|
logger = logging.getLogger("rf4.api")
|
|
app = FastAPI(title="RF4 Spotter API", version="0.1.0")
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=settings.cors_origins,
|
|
allow_methods=["GET", "POST", "PATCH", "DELETE"],
|
|
allow_headers=["Authorization", "Content-Type"],
|
|
)
|
|
|
|
|
|
@app.middleware("http")
|
|
async def structured_request_log(request: Request, call_next):
|
|
request_id = uuid.uuid4().hex
|
|
started = time_module.perf_counter()
|
|
status_code = 500
|
|
try:
|
|
response = await call_next(request)
|
|
status_code = response.status_code
|
|
response.headers["X-Request-ID"] = request_id
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
|
|
response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
|
|
response.headers["X-Frame-Options"] = "DENY"
|
|
response.headers["Cross-Origin-Opener-Policy"] = "same-origin"
|
|
if request.url.path.startswith("/api/v1/admin/") or request.url.path == "/api/v1/catch-reports":
|
|
response.headers["Cache-Control"] = "no-store"
|
|
if settings.deployment_environment == "production":
|
|
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
|
|
return response
|
|
except Exception as exc:
|
|
logger.error("request failed", extra={"request_id": request_id, "error_type": type(exc).__name__})
|
|
raise
|
|
finally:
|
|
logger.log(
|
|
logging.DEBUG if request.url.path in {"/health", "/ready"} else logging.INFO,
|
|
"request completed",
|
|
extra={
|
|
"request_id": request_id, "method": request.method,
|
|
"path": request.url.path, "status_code": status_code,
|
|
"duration_ms": round((time_module.perf_counter() - started) * 1000, 2),
|
|
},
|
|
)
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/ready")
|
|
def ready(db: Db) -> JSONResponse:
|
|
is_ready, components = readiness_report(
|
|
db, storage_client(), import_required=settings.official_import_required,
|
|
import_interval_seconds=settings.import_interval_seconds,
|
|
community_import_interval_seconds=settings.community_import_interval_seconds,
|
|
)
|
|
return JSONResponse(
|
|
status_code=200 if is_ready else 503,
|
|
content={"status": "ready" if is_ready else "not_ready", "version": settings.app_version, "revision": settings.app_revision, "components": components},
|
|
)
|
|
|
|
|
|
app.include_router(catalog_router)
|
|
app.include_router(media_router)
|
|
app.include_router(activity_router)
|
|
app.include_router(public_data_router)
|
|
app.include_router(admin_router)
|
|
app.include_router(submissions_router)
|
|
|
|
|
|
def _check_rate_limit(request: Request, db: Session) -> None:
|
|
check_rate_limit(request, db, settings)
|