mimic-big/backend/src/mimic/app.py

"""Flask application factory."""

from __future__ import annotations

from datetime import timedelta
from typing import cast

from flask import Flask, jsonify
from flask.typing import ResponseReturnValue
from werkzeug.exceptions import HTTPException

from mimic.api import register_blueprints
from mimic.auth.identity import load_user
from mimic.config import Settings, get_settings
from mimic.extensions import db, login_manager, migrate, socketio
from mimic.logging import configure_logging

# HTTP status → stable snake_case error code surfaced in the JSON envelope.
# Anything not listed falls back to "http_error".
_HTTP_ERROR_CODES: dict[int, str] = {
    400: "bad_request",
    401: "not_authenticated",
    403: "forbidden",
    404: "not_found",
    405: "method_not_allowed",
    409: "conflict",
    415: "unsupported_media_type",
    422: "validation_error",
    429: "rate_limited",
    500: "internal_error",
    503: "service_unavailable",
}


def create_app(settings: Settings | None = None) -> Flask:
    settings = settings or get_settings()
    configure_logging(settings.log_level, as_json=settings.log_json)

    app = Flask(__name__)
    # `strict_slashes=False` means routes match with or without the trailing
    # slash. Cross-origin clients keep their session cookie either way (a
    # 308 redirect could drop it on some browsers).
    app.url_map.strict_slashes = False
    app.config.update(
        SECRET_KEY=settings.secret_key.get_secret_value(),
        SQLALCHEMY_DATABASE_URI=settings.database_url,
        SQLALCHEMY_TRACK_MODIFICATIONS=False,
        SESSION_COOKIE_SECURE=settings.session_cookie_secure,
        SESSION_COOKIE_SAMESITE=settings.session_cookie_samesite,
        SESSION_COOKIE_HTTPONLY=True,
        PERMANENT_SESSION_LIFETIME=timedelta(minutes=settings.session_lifetime_minutes),
        MIMIC_SETTINGS=settings,
    )

    db.init_app(app)
    migrate.init_app(app, db, directory="src/mimic/db/migrations")
    login_manager.init_app(app)
    login_manager.user_loader(load_user)

    @login_manager.unauthorized_handler  # type: ignore[untyped-decorator]
    def _unauthorized() -> ResponseReturnValue:
        # API returns JSON; never redirect to a login page.
        return (
            jsonify({"error": "not_authenticated", "message": "no active session"}),
            401,
        )

    @app.errorhandler(HTTPException)
    def _json_http_error(exc: HTTPException) -> ResponseReturnValue:
        """Serialize every aborted request as the uniform JSON envelope.

        `flask.abort()` defaults to a Werkzeug HTML page; without this handler
        the contract documented in docs/api.md would only hold for 401s.
        """
        status = exc.code or 500
        # The Werkzeug type stub pins `description` to `str | None`, but
        # `flask.abort(..., description=<list|dict>)` legally smuggles richer
        # payloads through (we use this for Pydantic `errors()` on 422). Cast
        # to `object` so the runtime type-narrowing below is type-checked.
        description = cast(object, exc.description)
        message: str
        details: object | None = None
        if isinstance(description, str):
            message = description
        elif isinstance(description, list | dict):
            # Pydantic `exc.errors()` flows through `abort(422, description=...)`
            # as a list; keep it under `details` so the client can map per-field.
            message = "request failed"
            details = description
        else:
            message = exc.name or "request failed"
        body: dict[str, object] = {
            "error": _HTTP_ERROR_CODES.get(status, "http_error"),
            "message": message,
        }
        if details is not None:
            body["details"] = details
        return jsonify(body), status

    socketio.init_app(
        app,
        cors_allowed_origins=settings.cors_origins or "*",
        async_mode="gevent",
    )

    _enable_cors_in_dev(app, settings)
    register_blueprints(app)

    @app.get("/healthz")
    def healthz() -> ResponseReturnValue:
        return jsonify(status="ok"), 200

    return app


def _enable_cors_in_dev(app: Flask, settings: Settings) -> None:
    """Dev-only CORS for the Vite frontend on http://localhost:5173.

    In production, the reverse proxy (Caddy + same-origin) terminates this
    concern; enabling CORS there would expand the CSRF surface for no benefit.
    """
    if settings.env != "development":
        return
    if not settings.cors_origins:
        return
    from flask_cors import CORS  # noqa: PLC0415 — keeps the prod import path lean

    CORS(
        app,
        resources={r"/api/*": {"origins": settings.cors_origins}},
        supports_credentials=True,
        methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
        allow_headers=["Content-Type", "X-Requested-With"],
        max_age=600,
    )