feat(backend): add FastAPI app entry point
This commit is contained in:
124
backend/src/main.py
Normal file
124
backend/src/main.py
Normal file
@@ -0,0 +1,124 @@
|
||||
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"})
|
||||
Reference in New Issue
Block a user