Loading...
Loading...
FastAPI patterns for async APIs, dependency injection, Pydantic request and response models, OpenAPI docs, tests, security, and production readiness.
npx skill4agent add affaan-m/everything-claude-code fastapi-patternsmain.pyschemas/dependencies.pyservices/crud/tests/response_modelapp/
|-- main.py
|-- config.py
|-- dependencies.py
|-- exceptions.py
|-- api/
| `-- routes/
| |-- users.py
| `-- health.py
|-- core/
| |-- security.py
| `-- middleware.py
|-- db/
| |-- session.py
| `-- crud.py
|-- models/
|-- schemas/
`-- tests/from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.routes import health, users
from app.config import settings
from app.db.session import close_db, init_db
from app.exceptions import register_exception_handlers
@asynccontextmanager
async 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()allow_origins=["*"]allow_credentials=Truefrom datetime import datetime
from typing import Annotated
from 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: datetimefrom collections.abc import AsyncIterator
from uuid import UUID
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.security import decode_token
from app.db.session import session_factory
from 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 userfrom fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_current_user, get_db
from app.models.user import User
from 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()httpx.AsyncClientrequestsfrom fastapi import FastAPI, Request
from 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}},
)app.openapifrom fastapi import FastAPI
from 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_openapiDependsimport pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_db
from app.main import create_app
@pytest.fixture
async 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()argon2-cffibcryptcreate_appUserCreateUserUpdateUserResponseget_dbapp.openapi = custom_openapifastapi-reviewer/fastapi-reviewpython-patternspython-testingapi-design