fin-tracker/backend/src/fintracker/api/app.py

"""FastAPI application factory."""

from __future__ import annotations

import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.openapi.utils import get_openapi
from fastapi.routing import APIRoute

from fintracker import __version__
from fintracker.api.errors import install_error_handlers
from fintracker.api.routers import (
    accounts,
    analytics,
    auth,
    cashflow,
    categories,
    events,
    health,
    instruments,
    links,
    metrics,
    networth,
    rules,
    sync,
    transactions,
)
from fintracker.api.web import mount_web
from fintracker.config import get_settings
from fintracker.db import reset_engine

log = logging.getLogger(__name__)

API_PREFIX = "/api/v1"


def _operation_id(route: APIRoute) -> str:
    # "auth_login" instead of "login_api_v1_auth_login_post": readable Dart method names
    tag = route.tags[0] if route.tags else "default"
    return f"{tag}_{route.name}"


@asynccontextmanager
async def _lifespan(app: FastAPI) -> AsyncIterator[None]:
    settings = get_settings()
    if settings.is_dev_secret:
        log.warning("JWT_SECRET is the insecure development default")
    yield
    await reset_engine()


def create_app() -> FastAPI:
    settings = get_settings()
    app = FastAPI(
        title="fin-tracker",
        version=__version__,
        lifespan=_lifespan,
        generate_unique_id_function=_operation_id,
        docs_url=f"{API_PREFIX}/docs",
        openapi_url=f"{API_PREFIX}/openapi.json",
        redoc_url=None,
    )
    if settings.cors_origin_list:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=settings.cors_origin_list,
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )
    install_error_handlers(app)
    _install_openapi(app)
    app.include_router(health.router, prefix=API_PREFIX)
    app.include_router(auth.router, prefix=API_PREFIX)
    app.include_router(sync.router, prefix=API_PREFIX)
    app.include_router(accounts.router, prefix=API_PREFIX)
    app.include_router(categories.router, prefix=API_PREFIX)
    app.include_router(transactions.router, prefix=API_PREFIX)
    app.include_router(rules.router, prefix=API_PREFIX)
    app.include_router(networth.router, prefix=API_PREFIX)
    app.include_router(cashflow.router, prefix=API_PREFIX)
    app.include_router(analytics.router, prefix=API_PREFIX)
    app.include_router(events.router, prefix=API_PREFIX)
    app.include_router(instruments.router, prefix=API_PREFIX)
    app.include_router(links.router, prefix=API_PREFIX)
    app.include_router(metrics.router, prefix=API_PREFIX)
    if settings.web_dir is not None:
        mount_web(app, settings.web_dir, API_PREFIX)
    return app


PROBLEM_SCHEMA = {
    "title": "Problem",
    "type": "object",
    "description": "RFC 7807 error body (application/problem+json)",
    "required": ["status", "title"],
    "properties": {
        "status": {"type": "integer"},
        "title": {"type": "string"},
        "detail": {"type": "string"},
        "errors": {"type": "array", "items": {}},
    },
}


def _install_openapi(app: FastAPI) -> None:
    """Post-process the schema: every error is a Problem, never FastAPI's anyOf-heavy
    HTTPValidationError (which client generators cannot represent)."""

    def custom_openapi() -> dict[str, Any]:
        if app.openapi_schema:
            return app.openapi_schema
        schema = get_openapi(
            title=app.title,
            version=app.version,
            routes=app.routes,
            servers=[{"url": "/"}],
        )
        components = schema.setdefault("components", {}).setdefault("schemas", {})
        components.pop("HTTPValidationError", None)
        components.pop("ValidationError", None)
        components["Problem"] = PROBLEM_SCHEMA
        problem_ref = {
            "description": "Error",
            "content": {
                "application/problem+json": {"schema": {"$ref": "#/components/schemas/Problem"}}
            },
        }
        for path_item in schema.get("paths", {}).values():
            for op in path_item.values():
                responses = op.get("responses", {})
                responses.pop("422", None)
                responses["default"] = problem_ref
        app.openapi_schema = schema
        return schema

    app.openapi = custom_openapi  # type: ignore[method-assign]