Use when building FastAPI applications, implementing async endpoints, setting up Pydantic schemas, working with SQLAlchemy, or writing pytest tests for Python backend services.
Python patterns for building backend services with FastAPI.
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
# 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"}
# 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()
# 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
# 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())
# 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()
# 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)
# 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")
# 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
# 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": "john@example.com",
"password": "password123",
})
assert response.status_code == 201
data = response.json()
assert data["name"] == "John Doe"
assert data["email"] == "john@example.com"
@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
