63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
"""FastAPI api."""
|
|
|
|
# pyright: reportUnusedFunction=false
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import (
|
|
FastAPI,
|
|
Request,
|
|
status,
|
|
)
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
|
|
|
|
from .models import init_db_async
|
|
from .backend_api import get_backend_api
|
|
from .db import get_async_engine
|
|
|
|
from .settings import BackendSettings
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
def init_backend_app(settings: BackendSettings) -> FastAPI:
|
|
"""Initialize backend app."""
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_app: FastAPI):
|
|
"""Create database before starting the server."""
|
|
LOG.debug("Running lifespan")
|
|
engine = get_async_engine(settings.async_db_url)
|
|
await init_db_async(engine)
|
|
yield
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
app.include_router(get_backend_api(settings))
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(
|
|
request: Request, exc: RequestValidationError
|
|
):
|
|
return JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
|
|
)
|
|
|
|
@app.get("/health")
|
|
async def get_health() -> JSONResponse:
|
|
"""Provide simple health check."""
|
|
return JSONResponse(
|
|
status_code=status.HTTP_200_OK, content=jsonable_encoder({"status": "LIVE"})
|
|
)
|
|
|
|
return app
|
|
|
|
|
|
def create_backend_app(settings: BackendSettings) -> FastAPI:
|
|
"""Create the backend app."""
|
|
|
|
return init_backend_app(settings)
|