home / skills / madappgang / claude-code / python
This skill helps you build robust FastAPI backends with async endpoints, Pydantic schemas, SQLAlchemy models, and tests.
npx playbooks add skill madappgang/claude-code --skill pythonReview the files below or copy the command above to add this skill to your agents.
---
name: python
version: 1.0.0
description: Use when building FastAPI applications, implementing async endpoints, setting up Pydantic schemas, working with SQLAlchemy, or writing pytest tests for Python backend services.
keywords:
- Python
- FastAPI
- Pydantic
- SQLAlchemy
- async
- pytest
- backend
- API
plugin: dev
updated: 2026-01-20
---
# Python Backend Patterns
## Overview
Python patterns for building backend services with FastAPI.
## Project Structure
```
project/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI app
│ ├── config.py # Configuration
│ ├── dependencies.py # Dependency injection
│ ├── routers/ # API routes
│ │ ├── __init__.py
│ │ └── users.py
│ ├── services/ # Business logic
│ ├── repositories/ # Data access
│ ├── models/ # SQLAlchemy models
│ ├── schemas/ # Pydantic schemas
│ └── utils/ # Utilities
├── tests/ # Test files
├── migrations/ # Alembic migrations
├── pyproject.toml
└── requirements.txt
```
## FastAPI Application
### Main Application
```python
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from app.config import settings
from app.routers import users, auth
from app.database import engine, Base
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# Shutdown
await engine.dispose()
app = FastAPI(
title="My API",
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/users", tags=["users"])
@app.get("/health")
async def health():
return {"status": "ok"}
```
### Configuration
```python
# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
database_url: str = "postgresql+asyncpg://localhost/app"
redis_url: str = "redis://localhost:6379"
secret_key: str = "your-secret-key"
access_token_expire_minutes: int = 30
cors_origins: list[str] = ["http://localhost:3000"]
class Config:
env_file = ".env"
@lru_cache
def get_settings() -> Settings:
return Settings()
settings = get_settings()
```
## Pydantic Schemas
```python
# app/schemas/user.py
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Optional
class UserBase(BaseModel):
name: str = Field(..., min_length=2, max_length=100)
email: EmailStr
class UserCreate(UserBase):
password: str = Field(..., min_length=8)
class UserUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=2, max_length=100)
email: Optional[EmailStr] = None
class UserResponse(UserBase):
id: str
created_at: datetime
class Config:
from_attributes = True
class PaginatedResponse(BaseModel):
items: list[UserResponse]
total: int
page: int
page_size: int
```
## SQLAlchemy Models
```python
# app/models/user.py
from sqlalchemy import Column, String, DateTime, Boolean
from sqlalchemy.sql import func
from app.database import Base
import uuid
class User(Base):
__tablename__ = "users"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
name = Column(String(100), nullable=False)
email = Column(String(255), unique=True, nullable=False, index=True)
password_hash = Column(String(255), nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
```
## Async Database
```python
# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from app.config import settings
engine = create_async_engine(settings.database_url, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
async def get_db():
async with async_session() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
```
## Repository Pattern
```python
# app/repositories/user.py
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
class UserRepository:
def __init__(self, session: AsyncSession):
self.session = session
async def find_by_id(self, id: str) -> User | None:
result = await self.session.execute(
select(User).where(User.id == id)
)
return result.scalar_one_or_none()
async def find_by_email(self, email: str) -> User | None:
result = await self.session.execute(
select(User).where(User.email == email)
)
return result.scalar_one_or_none()
async def find_all(self, page: int = 1, page_size: int = 20) -> tuple[list[User], int]:
offset = (page - 1) * page_size
# Get items
result = await self.session.execute(
select(User)
.order_by(User.created_at.desc())
.offset(offset)
.limit(page_size)
)
items = result.scalars().all()
# Get total count
count_result = await self.session.execute(
select(func.count()).select_from(User)
)
total = count_result.scalar()
return list(items), total
async def create(self, data: UserCreate, password_hash: str) -> User:
user = User(
name=data.name,
email=data.email,
password_hash=password_hash,
)
self.session.add(user)
await self.session.flush()
return user
async def update(self, user: User, data: UserUpdate) -> User:
for key, value in data.model_dump(exclude_unset=True).items():
setattr(user, key, value)
await self.session.flush()
return user
async def delete(self, user: User) -> None:
await self.session.delete(user)
```
## API Routes
```python
# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.repositories.user import UserRepository
from app.services.user import UserService
from app.schemas.user import UserCreate, UserUpdate, UserResponse, PaginatedResponse
from app.dependencies import get_current_user
router = APIRouter()
def get_user_service(db: AsyncSession = Depends(get_db)) -> UserService:
return UserService(UserRepository(db))
@router.get("", response_model=PaginatedResponse)
async def list_users(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
service: UserService = Depends(get_user_service),
):
users, total = await service.find_all(page, page_size)
return PaginatedResponse(
items=users,
total=total,
page=page,
page_size=page_size,
)
@router.get("/{id}", response_model=UserResponse)
async def get_user(
id: str,
service: UserService = Depends(get_user_service),
):
user = await service.find_by_id(id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
data: UserCreate,
service: UserService = Depends(get_user_service),
):
return await service.create(data)
@router.patch("/{id}", response_model=UserResponse)
async def update_user(
id: str,
data: UserUpdate,
service: UserService = Depends(get_user_service),
current_user: User = Depends(get_current_user),
):
user = await service.update(id, data)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
id: str,
service: UserService = Depends(get_user_service),
current_user: User = Depends(get_current_user),
):
success = await service.delete(id)
if not success:
raise HTTPException(status_code=404, detail="User not found")
```
## Authentication
```python
# app/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from app.config import settings
from app.models.user import User
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db),
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
user_id: str = payload.get("sub")
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
repo = UserRepository(db)
user = await repo.find_by_id(user_id)
if user is None:
raise credentials_exception
return user
```
## Testing
```python
# tests/test_users.py
import pytest
from httpx import AsyncClient
from app.main import app
from app.database import get_db, async_session
@pytest.fixture
async def client():
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.fixture
async def db():
async with async_session() as session:
yield session
await session.rollback()
@pytest.mark.asyncio
async def test_list_users(client: AsyncClient):
response = await client.get("/api/users")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
response = await client.post("/api/users", json={
"name": "John Doe",
"email": "[email protected]",
"password": "password123",
})
assert response.status_code == 201
data = response.json()
assert data["name"] == "John Doe"
assert data["email"] == "[email protected]"
@pytest.mark.asyncio
async def test_create_user_validation(client: AsyncClient):
response = await client.post("/api/users", json={
"name": "J", # Too short
"email": "invalid", # Invalid email
})
assert response.status_code == 422
```
---
*Python FastAPI patterns for backend development*
This skill provides concise patterns and examples for building Python backend services with FastAPI, async SQLAlchemy, Pydantic schemas, repository/service layers, and pytest tests. It focuses on practical project layout and interoperable components to accelerate reliable API development. Use it to implement async endpoints, schema validation, database access, and authentication with clear conventions.
The skill describes a recommended project structure, a FastAPI app lifecycle using an asynccontextmanager, and how to wire middleware and routers. It shows how to define Pydantic request/response schemas, async SQLAlchemy models and sessions, a repository pattern for data access, service-layer usage in routes, JWT-based dependency authentication, and basic pytest-based integration tests. Examples include pagination, create/update flows, and transactional session handling.
How should I structure Pydantic models for partial updates?
Use a separate Update schema with Optional fields and min/max validations, then apply model_dump(exclude_unset=True) to update only provided fields.
How do I manage async DB sessions across requests?
Provide a get_db dependency that yields an AsyncSession, commits on success, rolls back on exceptions, and ensures the session is closed in finally.
Should I return ORM models directly from endpoints?
Return Pydantic response models configured with from_attributes=True to serialize ORM instances; avoid exposing internal fields like password_hash.