--- name: python-fastapi-patterns description: FastAPI patterns — async routes, dependency injection, Pydantic, SQLAlchemy, Alembic, background tasks --- # Python FastAPI Patterns ## Project Structure ``` app/ ├── main.py # FastAPI app ├── config.py # Settings (pydantic-settings) ├── database.py # Database session ├── dependencies.py # Shared dependencies (auth, db, pagination) ├── models/ # SQLAlchemy models │ ├── product.py │ ├── order.py │ └── user.py ├── schemas/ # Pydantic schemas │ ├── product.py │ ├── order.py │ └── auth.py ├── routers/ # API routers │ ├── products.py │ ├── orders.py │ ├── auth.py │ └── admin.py ├── services/ # Business logic │ ├── product_service.py │ ├── order_service.py │ └── auth_service.py ├── repositories/ # Data access │ ├── product_repository.py │ └── order_repository.py ├── middleware/ │ ├── auth.py │ └── logging.py ├── tasks/ # Background tasks │ └── notifications.py ├── exceptions.py # Custom exceptions ├── migrations/ # Alembic migrations │ ├── env.py │ └── versions/ └── tests/ ├── conftest.py ├── test_products.py └── test_auth.py ``` ## App Setup ```python # app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.config import settings from app.routers import products, orders, auth app = FastAPI( title=settings.APP_NAME, version=settings.APP_VERSION, docs_url='/api/docs', redoc_url='/api/redoc', ) app.add_middleware( CORSMiddleware, allow_origins=settings.ALLOWED_ORIGINS, allow_credentials=True, allow_methods=['*'], allow_headers=['*'], ) app.include_router(products.router, prefix='/api/v1', tags=['products']) app.include_router(orders.router, prefix='/api/v1', tags=['orders']) app.include_router(auth.router, prefix='/api/v1/auth', tags=['auth']) @app.get('/health') async def health_check(): return {'status': 'ok'} ``` ## Config ```python # app/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): APP_NAME: str = 'My API' APP_VERSION: str = '1.0.0' DATABASE_URL: str = 'postgresql+asyncpg://user:pass@localhost/db' SECRET_KEY: str = 'change-me-in-production' ALLOWED_ORIGINS: list[str] = ['http://localhost:3000'] ACCESS_TOKEN_EXPIRE_MINUTES: int = 30 model_config = {'env_file': '.env', 'extra': 'ignore'} settings = Settings() ``` ## Database ```python # app/database.py from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from app.config import settings engine = create_async_engine(settings.DATABASE_URL, echo=False) AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def get_db() -> AsyncSession: async with AsyncSessionLocal() as session: yield session ``` ## Pydantic Schemas ```python # app/schemas/product.py from pydantic import BaseModel, Field class ProductBase(BaseModel): name: str = Field(min_length=1, max_length=255) price: float = Field(gt=0) category_id: int class ProductCreate(ProductBase): description: str | None = None class ProductUpdate(BaseModel): name: str | None = Field(None, min_length=1, max_length=255) price: float | None = Field(None, gt=0) class ProductResponse(ProductBase): id: int description: str | None slug: str model_config = {'from_attributes': True} class ProductList(BaseModel): data: list[ProductResponse] total: int page: int pages: int ``` ## Repository Pattern ```python # app/repositories/product_repository.py from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from app.models.product import Product class ProductRepository: def __init__(self, db: AsyncSession): self.db = db async def list(self, page=1, per_page=20, category_id=None, search=None): query = select(Product).where(Product.is_active.is_(True)) if category_id: query = query.where(Product.category_id == category_id) if search: query = query.where(Product.name.ilike(f'%{search}%')) total = await self.db.scalar(select(func.count()).select_from(query.subquery())) items = await self.db.scalars( query.order_by(Product.created_at.desc()) .offset((page - 1) * per_page) .limit(per_page) ) return items.all(), total async def get_by_id(self, product_id: int): return await self.db.get(Product, product_id) async def create(self, data: dict): product = Product(**data) self.db.add(product) await self.db.commit() await self.db.refresh(product) return product ``` ## Router Pattern (Thin) ```python # app/routers/products.py from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.schemas.product import ProductCreate, ProductResponse, ProductList from app.services.product_service import ProductService from app.dependencies import PaginationParams router = APIRouter() def get_service(db: AsyncSession = Depends(get_db)) -> ProductService: return ProductService(db) @router.get('/products', response_model=ProductList) async def list_products( pagination: PaginationParams = Depends(), category_id: int | None = None, search: str | None = None, service: ProductService = Depends(get_service), ): return await service.list( page=pagination.page, per_page=pagination.per_page, category_id=category_id, search=search, ) @router.post('/products', response_model=ProductResponse, status_code=201) async def create_product( data: ProductCreate, service: ProductService = Depends(get_service), ): return await service.create(data.model_dump()) @router.get('/products/{product_id}', response_model=ProductResponse) async def get_product(product_id: int, service: ProductService = Depends(get_service)): product = await service.get(product_id) if not product: raise HTTPException(status_code=404, detail='Product not found') return product ``` ## Authentication (JWT) ```python # app/services/auth_service.py from datetime import datetime, timedelta, timezone from jose import jwt, JWTError from passlib.context import CryptContext from app.config import settings pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto') class AuthService: @staticmethod def hash_password(password: str) -> str: return pwd_context.hash(password) @staticmethod def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) @staticmethod def create_access_token(user_id: int) -> str: expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) return jwt.encode({'sub': str(user_id), 'exp': expire}, settings.SECRET_KEY) @staticmethod def decode_token(token: str) -> dict: try: return jwt.decode(token, settings.SECRET_KEY, algorithms=['HS256']) except JWTError: raise ValueError('Invalid token') ``` ## Dependencies (Auth + Pagination) ```python # app/dependencies.py from fastapi import Depends, HTTPException, Header from app.services.auth_service import AuthService class PaginationParams: def __init__(self, page: int = 1, per_page: int = 20): self.page = max(1, page) self.per_page = min(100, max(1, per_page)) async def get_current_user(authorization: str = Header(...)): if not authorization.startswith('Bearer '): raise HTTPException(status_code=401, detail='Invalid token format') token = authorization[7:] try: payload = AuthService.decode_token(token) return int(payload['sub']) except (ValueError, KeyError): raise HTTPException(status_code=401, detail='Invalid or expired token') ``` ## Checklist - [ ] Async everywhere (routes, db, external calls) - [ ] Dependency injection for services and auth - [ ] Pydantic v2 schemas with `model_config = {'from_attributes': True}` - [ ] Repository pattern for data access - [ ] Service layer for business logic - [ ] Alembic for database migrations - [ ] JWT for authentication - [ ] CORS properly configured - [ ] pytest-asyncio for async tests - [ ] `alembic upgrade head` before server start - [ ] Health check endpoint for monitoring