3deaa5ef40
Complete Flask → FastAPI migration with: - FastAPI app with session auth, Azure AD SSO, rate limiting - SQLite-backed session store (survives restarts) - Bulk AI metadata generation with SSE progress - Admin panel (user management, audit log, AI usage) - Subpath deployment support (ROOT_PATH config) - Docker + deploy.sh for production deployment - Test suite (auth, upload, templates, imports, admin, sessions) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
"""FastAPI application factory with lifespan management."""
|
|
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, Request, Depends
|
|
from fastapi.exceptions import HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from slowapi import _rate_limit_exceeded_handler
|
|
from slowapi.errors import RateLimitExceeded
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
|
|
from .config import get_settings
|
|
from .dependencies import init_dependencies, get_current_user
|
|
from .security import limiter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
"""Startup/shutdown lifecycle."""
|
|
settings = get_settings()
|
|
init_dependencies(settings)
|
|
logger.info(f"{settings.APP_NAME} v{settings.APP_VERSION} starting")
|
|
yield
|
|
logger.info("Shutting down")
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
settings = get_settings()
|
|
|
|
app = FastAPI(
|
|
title=settings.APP_NAME,
|
|
version=settings.APP_VERSION,
|
|
root_path=settings.ROOT_PATH,
|
|
docs_url="/docs" if settings.DEBUG else None,
|
|
redoc_url=None,
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
app.state.limiter = limiter
|
|
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
|
|
|
|
# CORS — same origin only (restrict in production)
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=[settings.REDIRECT_URI.rsplit("/", 1)[0]] if not settings.DEBUG else ["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# Session middleware (cookie-based)
|
|
app.add_middleware(
|
|
SessionMiddleware,
|
|
secret_key=settings.SECRET_KEY,
|
|
session_cookie="oliver_session",
|
|
max_age=settings.SESSION_EXPIRE_HOURS * 3600,
|
|
same_site="lax",
|
|
https_only=settings.HTTPS_ONLY,
|
|
)
|
|
|
|
# Static files
|
|
project_root = Path(__file__).parent.parent
|
|
static_dir = project_root / "static"
|
|
if static_dir.exists():
|
|
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
|
|
|
|
# Templates
|
|
templates = Jinja2Templates(directory=settings.TEMPLATES_DIR)
|
|
|
|
# Register routers
|
|
from .routers import auth as auth_router
|
|
from .routers import upload as upload_router
|
|
from .routers import metadata as metadata_router
|
|
from .routers import templates as templates_router
|
|
from .routers import imports as imports_router
|
|
from .routers import downloads as downloads_router
|
|
from .routers import sse as sse_router
|
|
from .routers import admin as admin_router
|
|
|
|
auth_router.set_templates(templates)
|
|
admin_router.set_templates(templates)
|
|
app.include_router(auth_router.router)
|
|
app.include_router(upload_router.router)
|
|
app.include_router(metadata_router.router)
|
|
app.include_router(templates_router.router)
|
|
app.include_router(imports_router.router)
|
|
app.include_router(downloads_router.router)
|
|
app.include_router(sse_router.router)
|
|
app.include_router(admin_router.router)
|
|
|
|
# Main page
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def index(request: Request, user=Depends(get_current_user)):
|
|
return templates.TemplateResponse(
|
|
"index.html",
|
|
{
|
|
"request": request,
|
|
"username": user["username"],
|
|
"docker_mode": settings.DOCKER_MODE,
|
|
},
|
|
)
|
|
|
|
# Redirect unauthenticated users to login
|
|
@app.exception_handler(HTTPException)
|
|
async def http_exception_handler(request: Request, exc: HTTPException):
|
|
if exc.status_code == 401:
|
|
root = request.scope.get("root_path", "")
|
|
return RedirectResponse(url=f"{root}/login?next={request.url.path}", status_code=302)
|
|
# Re-raise other HTTP exceptions as JSON
|
|
from fastapi.responses import JSONResponse
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={"detail": exc.detail},
|
|
)
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|