APAW/.kilo/skills/python-fastapi-patterns/SKILL.md

name, description

name	description
python-fastapi-patterns	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

# 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

# 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

# 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

# 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

# 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)

# 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)

# 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)

# 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