125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
import logging
|
|
import time
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.responses import Response
|
|
|
|
from src.api.routes import categories, patterns, questions, stats
|
|
from src.config import get_settings
|
|
from src.db.database import async_session_factory
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
settings = get_settings()
|
|
|
|
app = FastAPI(
|
|
title=settings.app_name,
|
|
version=settings.app_version,
|
|
description="API for coding interview preparation with curated educational content",
|
|
)
|
|
|
|
|
|
class RequestLoggingMiddleware(BaseHTTPMiddleware):
|
|
"""Middleware to log incoming requests with timing."""
|
|
|
|
async def dispatch(
|
|
self,
|
|
request: Request,
|
|
call_next: Callable[[Request], Awaitable[Response]],
|
|
) -> Response:
|
|
"""Log request method, path, status, and duration."""
|
|
start_time = time.perf_counter()
|
|
response = await call_next(request)
|
|
duration_ms = (time.perf_counter() - start_time) * 1000
|
|
|
|
logger.info(
|
|
"%s %s - %d (%.2fms)",
|
|
request.method,
|
|
request.url.path,
|
|
response.status_code,
|
|
duration_ms,
|
|
)
|
|
return response
|
|
|
|
|
|
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
|
|
"""Middleware to add security headers to all responses."""
|
|
|
|
async def dispatch(
|
|
self,
|
|
request: Request,
|
|
call_next: Callable[[Request], Awaitable[Response]],
|
|
) -> Response:
|
|
"""Add security headers to response."""
|
|
response = await call_next(request)
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["X-Frame-Options"] = "DENY"
|
|
response.headers["X-XSS-Protection"] = "1; mode=block"
|
|
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
|
|
response.headers["Content-Security-Policy"] = "default-src 'self'; frame-ancestors 'none'"
|
|
return response
|
|
|
|
|
|
app.add_middleware(RequestLoggingMiddleware)
|
|
app.add_middleware(SecurityHeadersMiddleware)
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=settings.cors_origins,
|
|
allow_credentials=True,
|
|
allow_methods=["GET", "HEAD", "OPTIONS", "POST"],
|
|
allow_headers=["Content-Type", "Accept"],
|
|
)
|
|
|
|
|
|
@app.exception_handler(SQLAlchemyError)
|
|
async def sqlalchemy_exception_handler(
|
|
request: Request, # noqa: ARG001
|
|
exc: SQLAlchemyError,
|
|
) -> JSONResponse:
|
|
"""Handle database errors."""
|
|
logger.exception("Database error: %s", exc)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"detail": "A database error occurred"},
|
|
)
|
|
|
|
|
|
@app.exception_handler(Exception)
|
|
async def generic_exception_handler(
|
|
request: Request, # noqa: ARG001
|
|
exc: Exception,
|
|
) -> JSONResponse:
|
|
"""Handle unexpected errors."""
|
|
logger.exception("Unexpected error: %s", exc)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"detail": "An unexpected error occurred"},
|
|
)
|
|
|
|
|
|
app.include_router(questions.router, prefix="/api/questions", tags=["questions"])
|
|
app.include_router(categories.router, prefix="/api/categories", tags=["categories"])
|
|
app.include_router(patterns.router, prefix="/api/patterns", tags=["patterns"])
|
|
app.include_router(stats.router, prefix="/api/stats", tags=["stats"])
|
|
|
|
|
|
@app.get("/api/health", response_model=None)
|
|
async def health_check() -> dict[str, str] | JSONResponse:
|
|
"""Health check endpoint with database verification."""
|
|
try:
|
|
async with async_session_factory() as session:
|
|
await session.execute(text("SELECT 1"))
|
|
return {"status": "healthy"}
|
|
except Exception:
|
|
logger.exception("Health check failed")
|
|
return JSONResponse(status_code=503, content={"status": "unhealthy"})
|