main.py

import os
from collections import defaultdict
from contextlib import asynccontextmanager
from http import HTTPStatus

from fastapi import APIRouter, FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import JSONResponse
from pydantic import ValidationError

from cms_backend.api.routes.account import router as account_router
from cms_backend.api.routes.auth import router as auth_router
from cms_backend.api.routes.books import router as books_router
from cms_backend.api.routes.collection import router as collection_router
from cms_backend.api.routes.config import router as config_router
from cms_backend.api.routes.events import router as events_router
from cms_backend.api.routes.healthcheck import router as healthcheck_router
from cms_backend.api.routes.http_errors import BadRequestError
from cms_backend.api.routes.staging import router as staging_router
from cms_backend.api.routes.titles import router as titles_router
from cms_backend.api.routes.warehouse import router as warehouse_router
from cms_backend.api.routes.zimfarm_notifications import (
    router as zimfarm_notification_router,
)
from cms_backend.api.routes.zimfarm_recipes import (
    router as zimfarm_recipe_router,
)
from cms_backend.context import Context
from cms_backend.db.exceptions import (
    RecordAlreadyExistsError,
    RecordDisabledError,
    RecordDoesNotExistError,
)
from cms_backend.utils.database import (
    check_if_schema_is_up_to_date,
    create_initial_account,
    upgrade_db_schema,
)


@asynccontextmanager
async def lifespan(_: FastAPI):
    if Context.alembic_upgrade_head_on_start:
        upgrade_db_schema()
    check_if_schema_is_up_to_date()
    create_initial_account()
    yield


def create_app(*, debug: bool = True):
    app = FastAPI(
        debug=debug,
        docs_url="/",
        title="CMS API",
        version="1.0.0",
        description="CMS API for managing titles, books, and other resources",
        lifespan=lifespan,
    )

    app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=5)
    if origins := os.getenv("ALLOWED_ORIGINS", None):
        app.add_middleware(
            CORSMiddleware,
            allow_origins=origins.split(","),
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )

    main_router = APIRouter(prefix="/v1")
    main_router.include_router(router=config_router)
    main_router.include_router(router=zimfarm_notification_router)
    main_router.include_router(router=zimfarm_recipe_router)
    main_router.include_router(router=healthcheck_router)
    main_router.include_router(router=titles_router)
    main_router.include_router(router=books_router)
    main_router.include_router(router=collection_router)
    main_router.include_router(router=events_router)
    main_router.include_router(router=auth_router)
    main_router.include_router(router=account_router)
    main_router.include_router(router=staging_router)
    main_router.include_router(router=warehouse_router)

    app.include_router(router=main_router)

    return app


app = create_app(debug=Context.debug)


@app.exception_handler(RequestValidationError)
async def request_validation_error_handler(_, exc: RequestValidationError):
    # transform the pydantic validation errors to a dictionary mapping
    # the field to the list of errors
    errors: dict[str | int, list[str]] = defaultdict(list)
    for err in exc.errors():
        loc = err["loc"]
        key = loc[-1] if loc else "root"  # fallback for model level errors
        errors[key].append(err["msg"])

    return JSONResponse(
        status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        content={
            "success": False,
            "message": "Input values failed constraints validation",
            "errors": errors,
        },
    )


@app.exception_handler(ValueError)
async def value_error_handler(_, exc: ValueError):
    return JSONResponse(
        status_code=HTTPStatus.BAD_REQUEST,
        content={"success": False, "message": exc.args[0]},
    )


@app.exception_handler(ValidationError)
async def validation_error_handler(_, exc: ValidationError):
    # transform the pydantic validation errors to a dictionary mapping
    # the field to the list of errors
    errors: dict[str | int, list[str]] = defaultdict(list)
    for err in exc.errors():
        loc = err["loc"]
        key = loc[-1] if loc else "root"  # fallback for model level errors
        errors[key].append(err["msg"])
    return JSONResponse(
        status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        content={
            "success": False,
            "message": "Input values failed constraints validation",
            "errors": errors,
        },
    )


@app.exception_handler(RecordDoesNotExistError)
async def record_does_not_exist_error_handler(_, exc: RecordDoesNotExistError):
    return JSONResponse(
        status_code=HTTPStatus.NOT_FOUND,
        content={"success": False, "message": exc.detail},
    )


@app.exception_handler(RecordDisabledError)
async def record_disabled_error_handler(_, exc: RecordDisabledError):
    return JSONResponse(
        status_code=HTTPStatus.SERVICE_UNAVAILABLE,
        content={"success": False, "message": exc.detail},
    )


@app.exception_handler(RecordAlreadyExistsError)
async def record_already_exists_error_handler(_, exc: RecordAlreadyExistsError):
    return JSONResponse(
        status_code=HTTPStatus.CONFLICT,
        content={"success": False, "message": exc.detail},
    )


@app.exception_handler(BadRequestError)
async def bad_request_error_handler(_, exc: BadRequestError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"success": False, "message": exc.detail, "errors": exc.errors},
    )


@app.exception_handler(HTTPException)
async def http_exception_handler(_, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code, content={"success": False, "message": exc.detail}
    )


@app.exception_handler(Exception)
async def generic_error_handler(_, __):  # pyright: ignore
    return JSONResponse(
        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
        content={"success": False, "message": "Internal server error"},
    )