Add core router and main application
Implement application factory pattern with clean main.py. Separate routers for each domain, centralized exception handling. Core Router (src/core/router.py): - Root endpoint (/) - Health check endpoint (/health) - Simple status responses - No prefix (mounted at root) Main Application (src/main.py): - create_application() factory function - Clean configuration-focused main.py - CORS middleware setup - Exception handler registration - Router registration with proper prefixes - Global config integration Application Architecture: - Application factory pattern for testability - Routers imported from separate controllers - Exception handlers in dedicated function - All routes cleanly separated by domain Exception Handling: - OpenAI-compatible error format - Custom AppException handler - Validation error handler (422) - Generic exception handler (500) Following Best Practices: - Separation of concerns - Factory pattern for DI - Clean main.py (config only) - Type hints throughout Status: Production-ready structure 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,38 @@
|
||||
"""
|
||||
Core router for health and root endpoints.
|
||||
"""
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
from src.core.config import config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(tags=["core"])
|
||||
|
||||
|
||||
@router.get("/health")
|
||||
async def health_check() -> dict[str, str]:
|
||||
"""
|
||||
Health check endpoint.
|
||||
|
||||
Returns:
|
||||
Health status
|
||||
"""
|
||||
return {"status": "healthy", "version": config.APP_VERSION}
|
||||
|
||||
|
||||
@router.get("/")
|
||||
async def root() -> dict[str, str]:
|
||||
"""
|
||||
Root endpoint.
|
||||
|
||||
Returns:
|
||||
API information
|
||||
"""
|
||||
return {
|
||||
"name": config.APP_NAME,
|
||||
"version": config.APP_VERSION,
|
||||
"docs": "/docs",
|
||||
}
|
||||
+157
@@ -0,0 +1,157 @@
|
||||
"""
|
||||
FastAPI application entry point.
|
||||
Following best practices from github.com/zhanymkanov/fastapi-best-practices
|
||||
|
||||
Main responsibilities:
|
||||
- Application configuration
|
||||
- Middleware setup
|
||||
- Exception handlers
|
||||
- Router registration
|
||||
- Lifecycle management
|
||||
"""
|
||||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import AsyncGenerator
|
||||
|
||||
from fastapi import FastAPI, Request, status
|
||||
from fastapi.exceptions import RequestValidationError
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from src.chat.router import router as chat_router
|
||||
from src.core.config import config
|
||||
from src.core.exceptions import AppException
|
||||
from src.core.router import router as core_router
|
||||
from src.models.router import router as models_router
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=config.LOG_LEVEL,
|
||||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
|
||||
"""
|
||||
Application lifespan manager.
|
||||
|
||||
Handles startup and shutdown logic.
|
||||
"""
|
||||
# Startup
|
||||
logger.info(f"Starting {config.APP_NAME} v{config.APP_VERSION}")
|
||||
logger.info(f"Environment: {config.ENVIRONMENT.value}")
|
||||
logger.info(f"Ollama host: {config.OLLAMA_HOST}")
|
||||
logger.info(f"Default model: {config.OLLAMA_DEFAULT_MODEL}")
|
||||
|
||||
yield
|
||||
|
||||
# Shutdown
|
||||
logger.info("Shutting down application")
|
||||
|
||||
|
||||
def create_application() -> FastAPI:
|
||||
"""
|
||||
Application factory.
|
||||
|
||||
Creates and configures the FastAPI application.
|
||||
Following best practice of using factory pattern.
|
||||
"""
|
||||
# Create app
|
||||
application = FastAPI(
|
||||
title=config.APP_NAME,
|
||||
version=config.APP_VERSION,
|
||||
lifespan=lifespan,
|
||||
debug=config.DEBUG,
|
||||
)
|
||||
|
||||
# Add middleware
|
||||
application.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=config.CORS_ORIGINS,
|
||||
allow_credentials=config.CORS_ALLOW_CREDENTIALS,
|
||||
allow_methods=config.CORS_ALLOW_METHODS,
|
||||
allow_headers=config.CORS_ALLOW_HEADERS,
|
||||
)
|
||||
|
||||
# Register exception handlers
|
||||
register_exception_handlers(application)
|
||||
|
||||
# Include routers
|
||||
application.include_router(core_router) # Health and root endpoints
|
||||
application.include_router(chat_router, prefix=config.API_PREFIX)
|
||||
application.include_router(models_router, prefix=config.API_PREFIX)
|
||||
|
||||
return application
|
||||
|
||||
|
||||
def register_exception_handlers(application: FastAPI) -> None:
|
||||
"""
|
||||
Register global exception handlers.
|
||||
|
||||
Provides consistent error responses compatible with OpenAI API.
|
||||
"""
|
||||
|
||||
@application.exception_handler(AppException)
|
||||
async def app_exception_handler(
|
||||
request: Request,
|
||||
exc: AppException,
|
||||
) -> JSONResponse:
|
||||
"""Handle custom application exceptions."""
|
||||
logger.error(
|
||||
f"Application error: {exc.message}",
|
||||
extra={"details": exc.details}
|
||||
)
|
||||
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content={
|
||||
"error": {
|
||||
"message": exc.message,
|
||||
"type": exc.__class__.__name__,
|
||||
"details": exc.details,
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
@application.exception_handler(RequestValidationError)
|
||||
async def validation_exception_handler(
|
||||
request: Request,
|
||||
exc: RequestValidationError,
|
||||
) -> JSONResponse:
|
||||
"""Handle Pydantic validation errors."""
|
||||
logger.error(f"Validation error: {exc.errors()}")
|
||||
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
content={
|
||||
"error": {
|
||||
"message": "Invalid request",
|
||||
"type": "invalid_request_error",
|
||||
"details": exc.errors(),
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
@application.exception_handler(Exception)
|
||||
async def general_exception_handler(
|
||||
request: Request,
|
||||
exc: Exception,
|
||||
) -> JSONResponse:
|
||||
"""Handle unexpected exceptions."""
|
||||
logger.exception("Unexpected error")
|
||||
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
content={
|
||||
"error": {
|
||||
"message": "An internal error occurred",
|
||||
"type": "internal_server_error",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# Create application instance
|
||||
app = create_application()
|
||||
Reference in New Issue
Block a user