"""Application factory, lifespan and error handling.""" from __future__ import annotations import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager from fastapi import FastAPI, Request, status from fastapi.responses import JSONResponse, Response from fastapi.staticfiles import StaticFiles from starlette.exceptions import HTTPException as StarletteHTTPException from lembas import __version__ from lembas.api import ( admin, admin_models, admin_users, auth, chats, folders, pages, preferences, ) from lembas.api.deps import RedirectToLogin, is_htmx, login_redirect from lembas.config import settings from lembas.db.session import init_db from lembas.web.templating import STATIC_DIR, render log = logging.getLogger("lembas") def configure_logging() -> None: logging.basicConfig( level=settings.log_level.upper(), format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", datefmt="%H:%M:%S", ) @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: configure_logging() settings.ensure_dirs() init_db() if settings.secret_key_is_ephemeral: log.warning( "No LEMBAS_SECRET_KEY set, so a temporary one was generated. Every " "restart will sign all users out and make stored API keys " "unreadable. Generate a permanent key with:\n" ' python -c "import secrets; print(secrets.token_urlsafe(48))"' ) log.info("LLeMbas %s starting on http://%s:%s", __version__, settings.host, settings.port) log.info("data directory: %s", settings.data_dir.resolve()) yield log.info("LLeMbas stopped") def create_app() -> FastAPI: app = FastAPI( title="LLeMbas", version=__version__, lifespan=lifespan, # The API is an implementation detail of the UI, not a product surface. docs_url="/api/docs" if settings.log_level == "debug" else None, redoc_url=None, ) app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") app.include_router(pages.router) app.include_router(auth.router) app.include_router(preferences.router) app.include_router(chats.router) app.include_router(folders.router) app.include_router(admin.router) app.include_router(admin_users.router) app.include_router(admin_models.router) register_error_handlers(app) return app def register_error_handlers(app: FastAPI) -> None: @app.exception_handler(RedirectToLogin) async def _not_signed_in(request: Request, exc: RedirectToLogin) -> Response: # An htmx request must not swap a login page into a fragment of the # chat UI, so tell the browser to navigate instead. if is_htmx(request): response = Response(status_code=status.HTTP_204_NO_CONTENT) response.headers["HX-Redirect"] = "/auth/login" return response return login_redirect(exc.next_url) @app.exception_handler(StarletteHTTPException) async def _http_error(request: Request, exc: StarletteHTTPException) -> Response: # JSON callers and htmx fragments want the bare status; humans loading a # page want a themed page they can navigate away from. wants_page = "text/html" in request.headers.get("accept", "") and not is_htmx(request) if not wants_page: return JSONResponse({"detail": exc.detail}, status_code=exc.status_code) return render( request, "error.html", { "status_code": exc.status_code, "detail": exc.detail, "flavour": ERROR_FLAVOUR.get(exc.status_code, ERROR_FLAVOUR[500]), }, status_code=exc.status_code, ) @app.exception_handler(Exception) async def _unhandled(request: Request, exc: Exception) -> Response: log.exception("unhandled error at %s", request.url.path) if is_htmx(request) or "text/html" not in request.headers.get("accept", ""): return JSONResponse({"detail": "Internal server error"}, status_code=500) return render( request, "error.html", {"status_code": 500, "detail": "Something went wrong.", "flavour": ERROR_FLAVOUR[500]}, status_code=500, ) # Flavour lives in error pages, empty states and theme names -- never in the # functional UI. See CLAUDE.md. ERROR_FLAVOUR = { 403: "Speak, friend, and enter. This door is not yours to open.", 404: "Not all those who wander are lost. This page, however, is.", 500: "The Road goes ever on, but this stretch of it has washed out.", } app = create_app()