Fusion
843683d579
Add logging infrastructure: - RequestLoggingMiddleware: logs all requests with method, path, status, timing - ExceptionLoggingMiddleware: catches and logs unhandled exceptions with stack traces - configure_logging(): structured logging with configurable level via LOG_LEVEL env var Add detailed auth flow logging: - Login initiation - Token exchange success/failure - JWKS fetch success/failure - Token verification - User lookup/creation - Database errors - Final response This enables tracing Internal Server Errors through the logs.
93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
import logging
|
|
import sys
|
|
import time
|
|
import traceback
|
|
from typing import Callable
|
|
|
|
from fastapi import Request, Response
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RequestLoggingMiddleware(BaseHTTPMiddleware):
|
|
"""Log all HTTP requests with timing and status codes."""
|
|
|
|
async def dispatch(self, request: Request, call_next: Callable) -> Response:
|
|
start_time = time.time()
|
|
client_host = request.client.host if request.client else "unknown"
|
|
|
|
# Log the incoming request
|
|
logger.info(
|
|
"→ Request: %s %s (client: %s)",
|
|
request.method,
|
|
request.url.path,
|
|
client_host,
|
|
)
|
|
|
|
try:
|
|
response = await call_next(request)
|
|
duration = time.time() - start_time
|
|
|
|
# Log the response
|
|
logger.info(
|
|
"← Response: %s %s → %d (%dms)",
|
|
request.method,
|
|
request.url.path,
|
|
response.status_code,
|
|
int(duration * 1000),
|
|
)
|
|
return response
|
|
|
|
except Exception as exc:
|
|
duration = time.time() - start_time
|
|
logger.error(
|
|
"✗ Error: %s %s → %s (%dms)\n%s",
|
|
request.method,
|
|
request.url.path,
|
|
type(exc).__name__,
|
|
int(duration * 1000),
|
|
traceback.format_exc(),
|
|
)
|
|
raise
|
|
|
|
|
|
class ExceptionLoggingMiddleware(BaseHTTPMiddleware):
|
|
"""Catch and log all unhandled exceptions."""
|
|
|
|
async def dispatch(self, request: Request, call_next: Callable) -> Response:
|
|
try:
|
|
return await call_next(request)
|
|
except Exception:
|
|
logger.critical(
|
|
"Unhandled exception in %s %s:\n%s",
|
|
request.method,
|
|
request.url.path,
|
|
traceback.format_exc(),
|
|
)
|
|
raise
|
|
|
|
|
|
def configure_logging(level: int = logging.INFO) -> None:
|
|
"""Configure structured logging for the application."""
|
|
formatter = logging.Formatter(
|
|
fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
|
|
# Console handler
|
|
console_handler = logging.StreamHandler(sys.stdout)
|
|
console_handler.setFormatter(formatter)
|
|
|
|
# Configure root logger
|
|
root_logger = logging.getLogger()
|
|
root_logger.setLevel(level)
|
|
root_logger.handlers = [console_handler]
|
|
|
|
# Set levels for specific loggers
|
|
logging.getLogger("uvicorn").setLevel(logging.WARNING)
|
|
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
|
|
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
|
|
|
|
logger.info("Logging configured at level %s", logging.getLevelName(level))
|