130 lines
No EOL
4.1 KiB
Python
130 lines
No EOL
4.1 KiB
Python
import logging
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI, Request, status
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
from fastapi.responses import RedirectResponse
|
|
from starlette.staticfiles import StaticFiles
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import init_db, AsyncSessionLocal
|
|
from app.modules.endpoints.repositories.endpoint_repository import EndpointRepository
|
|
from app.modules.endpoints.services.route_service import RouteManager
|
|
from app.core.middleware.auth_middleware import AuthMiddleware
|
|
from app.modules.admin.controller import router as admin_router
|
|
from app.modules.oauth2 import oauth_router
|
|
|
|
|
|
# Convert log level string to logging constant
|
|
log_level_map = {
|
|
"DEBUG": logging.DEBUG,
|
|
"INFO": logging.INFO,
|
|
"WARNING": logging.WARNING,
|
|
"ERROR": logging.ERROR,
|
|
"CRITICAL": logging.CRITICAL,
|
|
}
|
|
log_level = log_level_map.get(settings.log_level.upper(), logging.INFO)
|
|
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if settings.debug else log_level,
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
"""
|
|
Lifespan context manager for startup and shutdown events.
|
|
"""
|
|
# Startup
|
|
logger.info("Initializing database...")
|
|
await init_db()
|
|
|
|
# Use the route manager already attached to app.state
|
|
route_manager = app.state.route_manager
|
|
logger.info("Refreshing routes...")
|
|
await route_manager.refresh_routes()
|
|
|
|
logger.info("Application startup complete.")
|
|
yield
|
|
# Shutdown
|
|
logger.info("Application shutting down...")
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
"""
|
|
Factory function to create and configure the FastAPI application.
|
|
"""
|
|
app = FastAPI(
|
|
title=settings.title,
|
|
version=settings.version,
|
|
debug=settings.debug,
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
# Attach route manager and session factory to app.state before any request
|
|
route_manager = RouteManager(app, AsyncSessionLocal)
|
|
app.state.route_manager = route_manager
|
|
app.state.session_factory = AsyncSessionLocal
|
|
|
|
# Add authentication middleware for admin routes (must be after SessionMiddleware)
|
|
app.add_middleware(AuthMiddleware)
|
|
# Add session middleware (must be before AuthMiddleware, but add_middleware prepends)
|
|
app.add_middleware(
|
|
SessionMiddleware,
|
|
secret_key=settings.secret_key,
|
|
session_cookie=settings.session_cookie_name,
|
|
max_age=settings.session_max_age,
|
|
https_only=False,
|
|
same_site="lax",
|
|
)
|
|
|
|
|
|
|
|
# Mount static files (optional, for future)
|
|
# app.mount("/static", StaticFiles(directory="app/static"), name="static")
|
|
|
|
# Add a simple health check endpoint
|
|
@app.get("/health")
|
|
async def health_check():
|
|
return {"status": "healthy", "service": "mock-api"}
|
|
|
|
# Add endpoint to display request headers
|
|
@app.get("/headers")
|
|
async def get_headers(request: Request):
|
|
"""Return all HTTP headers from the request (excluding sensitive ones)."""
|
|
# Get all headers
|
|
all_headers = dict(request.headers)
|
|
|
|
# Filter out sensitive headers (case-insensitive)
|
|
sensitive = {"authorization", "cookie", "set-cookie", "x-api-key"}
|
|
filtered = {
|
|
k: v for k, v in all_headers.items()
|
|
if k.lower() not in sensitive
|
|
}
|
|
|
|
# Return headers with request metadata
|
|
return {
|
|
"headers": filtered,
|
|
"client_ip": request.client.host if request.client else None,
|
|
"method": request.method,
|
|
"url": str(request.url)
|
|
}
|
|
|
|
# Redirect root to Swagger documentation
|
|
@app.get("/")
|
|
async def root_redirect():
|
|
"""Redirect the root URL to Swagger documentation."""
|
|
return RedirectResponse(url="/docs", status_code=status.HTTP_302_FOUND)
|
|
|
|
# Include admin controller routes
|
|
app.include_router(admin_router)
|
|
# Include OAuth2 routes
|
|
app.include_router(oauth_router)
|
|
|
|
return app
|
|
|
|
|
|
# Create the application instance
|
|
app = create_app() |