Fastapi Patterns

from contextlib import asynccontextmanager
from fastapi import FastAPIfrom fastapi.middleware.cors import CORSMiddleware
from app.api.routes import health, usersfrom app.config import settingsfrom app.db.session import close_db, init_dbfrom app.exceptions import register_exception_handlers
@asynccontextmanagerasync def lifespan(app: FastAPI):    await init_db()    yield    await close_db()
def create_app() -> FastAPI:    app = FastAPI(        title=settings.api_title,        version=settings.api_version,        lifespan=lifespan,    )
    app.add_middleware(        CORSMiddleware,        allow_origins=settings.cors_origins,        allow_credentials=bool(settings.cors_origins),        allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],        allow_headers=["Authorization", "Content-Type"],    )
    register_exception_handlers(app)    app.include_router(health.router, prefix="/health", tags=["health"])    app.include_router(users.router, prefix="/api/v1/users", tags=["users"])    return app
app = create_app()

from datetime import datetimefrom typing import Annotatedfrom uuid import UUID
from pydantic import BaseModel, ConfigDict, EmailStr, Field
class UserBase(BaseModel):    email: EmailStr    full_name: Annotated[str, Field(min_length=1, max_length=100)]
class UserCreate(UserBase):    password: Annotated[str, Field(min_length=12, max_length=128)]
class UserUpdate(BaseModel):    email: EmailStr | None = None    full_name: Annotated[str | None, Field(min_length=1, max_length=100)] = None
class UserResponse(UserBase):    model_config = ConfigDict(from_attributes=True)
    id: UUID    created_at: datetime    updated_at: datetime

from collections.abc import AsyncIteratorfrom uuid import UUID
from fastapi import Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearerfrom sqlalchemy.ext.asyncio import AsyncSession
from app.core.security import decode_tokenfrom app.db.session import session_factoryfrom app.models.user import User
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
async def get_db() -> AsyncIterator[AsyncSession]:    async with session_factory() as session:        try:            yield session            await session.commit()        except Exception:            await session.rollback()            raise
async def get_current_user(    token: str = Depends(oauth2_scheme),    db: AsyncSession = Depends(get_db),) -> User:    payload = decode_token(token)    user_id = UUID(payload["sub"])    user = await db.get(User, user_id)    if user is None:        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")    return user

from fastapi import APIRouter, Depends, Queryfrom sqlalchemy import selectfrom sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_current_user, get_dbfrom app.models.user import Userfrom app.schemas.user import UserResponse
router = APIRouter()
@router.get("/", response_model=list[UserResponse])async def list_users(    limit: int = Query(default=50, ge=1, le=100),    offset: int = Query(default=0, ge=0),    db: AsyncSession = Depends(get_db),    current_user: User = Depends(get_current_user),):    result = await db.execute(        select(User).order_by(User.created_at.desc()).limit(limit).offset(offset)    )    return result.scalars().all()

from fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponse
class ApiError(Exception):    def __init__(self, status_code: int, code: str, message: str):        self.status_code = status_code        self.code = code        self.message = message
def register_exception_handlers(app: FastAPI) -> None:    @app.exception_handler(ApiError)    async def api_error_handler(request: Request, exc: ApiError):        return JSONResponse(            status_code=exc.status_code,            content={"error": {"code": exc.code, "message": exc.message}},        )

from fastapi import FastAPIfrom fastapi.openapi.utils import get_openapi
def install_openapi(app: FastAPI) -> None:    def custom_openapi():        if app.openapi_schema:            return app.openapi_schema        app.openapi_schema = get_openapi(            title="Service API",            version="1.0.0",            routes=app.routes,        )        return app.openapi_schema
    app.openapi = custom_openapi

import pytestfrom httpx import ASGITransport, AsyncClientfrom sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_dbfrom app.main import create_app
@pytest.fixtureasync def client(test_session: AsyncSession):    app = create_app()
    async def override_get_db():        yield test_session
    app.dependency_overrides[get_db] = override_get_db    async with AsyncClient(        transport=ASGITransport(app=app),        base_url="http://test",    ) as test_client:        yield test_client    app.dependency_overrides.clear()