REST APIs, validation, auth, middleware, SQLAlchemy, background tasks, testing & UV package manager
Frameworkuv is an extremely fast Python package manager and project manager written in Rust. It replaces pip, pip-tools, virtualenv, pyenv, and more.
# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh
# or
brew install uv
# ββ Project Management ββ
uv init myproject # create new project with pyproject.toml
uv init myproject --lib # library project
cd myproject
# ββ Virtual Environment ββ
uv venv # create .venv (auto-detected Python)
uv venv --python 3.12 # with specific Python version
uv venv myenv # custom name
source .venv/bin/activate # activate (macOS/Linux)
# .venv\Scripts\activate # Windows
# ββ Install Python versions ββ
uv python install 3.12 # install Python 3.12
uv python list # list available versions
uv python pin 3.12 # pin version for project
# ββ Package Management ββ
uv add fastapi # add dependency (updates pyproject.toml)
uv add "fastapi[standard]" # with extras
uv add --dev pytest ruff # dev dependencies
uv remove requests # remove dependency
uv sync # install all deps from lock file
uv lock # generate/update lock file
# ββ Run commands ββ
uv run python app.py # run in project venv
uv run fastapi dev # run FastAPI dev server
uv run pytest # run tests
uv run ruff check . # linting
# ββ pip-compatible mode ββ
uv pip install fastapi # like pip but 10-100x faster
uv pip install -r requirements.txt
uv pip compile requirements.in -o requirements.txt # lock deps
# ββ One-shot script execution ββ
uv run --with requests python script.py # temp install + run
uvx ruff check . # run tool without installing[project]
name = "myapi"
version = "0.1.0"
requires-python = ">= 3.12"
dependencies = [
"fastapi[standard]>=0.115",
"sqlalchemy>=2.0",
"pydantic>=2.0",
"alembic>=1.13",
]
[dependency-groups]
dev = [
"pytest>=8.0",
"httpx>=0.27",
"ruff>=0.8",
]
[tool.ruff]
line-length = 88
target-version = "py312"# Install
uv add "fastapi[standard]"
# app/main.py
from fastapi import FastAPI
app = FastAPI(
title="My API",
version="1.0.0",
docs_url="/docs", # Swagger UI
redoc_url="/redoc", # ReDoc
)
@app.get("/")
async def root():
return {"message": "Hello World"}
# Run
# uv run fastapi dev # dev mode (auto-reload)
# uv run fastapi run # production
# uv run uvicorn app.main:app --reload --port 8000from fastapi import FastAPI, HTTPException, status
from fastapi.responses import JSONResponse
@app.get("/users")
async def list_users():
return users
@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
return user
@app.put("/users/{user_id}")
async def update_user(user_id: int, user: UserUpdate):
return {"id": user_id, **user.model_dump()}
@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: int):
return None
# Error handling
@app.get("/users/{user_id}")
async def get_user(user_id: int):
if user_id not in db:
raise HTTPException(status_code=404, detail="User not found")
return db[user_id]
# Router (split into files)
from fastapi import APIRouter
router = APIRouter(prefix="/api/v1", tags=["users"])
@router.get("/users")
async def get_users(): ...
# In main.py:
app.include_router(router)from pydantic import BaseModel, Field, EmailStr, field_validator
from datetime import datetime
from typing import Optional
class UserCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
email: EmailStr
age: int = Field(..., ge=0, le=150)
role: str = "user"
tags: list[str] = []
@field_validator("name")
@classmethod
def name_must_not_be_empty(cls, v):
if not v.strip():
raise ValueError("Name cannot be empty")
return v.strip()
class UserResponse(BaseModel):
id: int
name: str
email: str
created_at: datetime
model_config = {"from_attributes": True} # read from ORM models
class UserUpdate(BaseModel):
name: Optional[str] = None
email: Optional[EmailStr] = None
age: Optional[int] = None
# Response model
@app.get("/users", response_model=list[UserResponse])
async def list_users(): ...from fastapi import Query, Path
from enum import Enum
# Path parameters
@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., gt=0, description="User ID")):
...
# Query parameters with validation
@app.get("/users")
async def list_users(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
search: str | None = None,
sort_by: str = Query("created_at", pattern="^(name|email|created_at)$"),
roles: list[str] = Query([]), # ?roles=admin&roles=user
):
...
# Enum for fixed choices
class SortOrder(Enum):
asc = "asc"
desc = "desc"
@app.get("/items")
async def list_items(order: SortOrder = SortOrder.asc): ...from fastapi import Depends
# Simple dependency
async def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users")
async def list_users(db: Session = Depends(get_db)):
return db.query(User).all()
# Dependency with parameters
def pagination(skip: int = 0, limit: int = 20):
return {"skip": skip, "limit": limit}
@app.get("/items")
async def list_items(pagination: dict = Depends(pagination)):
...from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_token(data: dict, expires: timedelta = timedelta(hours=1)):
payload = {**data, "exp": datetime.utcnow() + expires}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
async def get_current_user(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id = payload.get("sub")
except:
raise HTTPException(status_code=401, detail="Invalid token")
return get_user(user_id)
@app.get("/me")
async def read_me(user = Depends(get_current_user)):
return userfrom fastapi.middleware.cors import CORSMiddleware
import time
# CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Custom middleware
@app.middleware("http")
async def add_timing(request, call_next):
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
response.headers["X-Process-Time"] = str(duration)
return response# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
DATABASE_URL = "postgresql://user:pass@localhost/mydb"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
class Base(DeclarativeBase): pass
# models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship, Mapped, mapped_column
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(String(255), unique=True)
posts: Mapped[list["Post"]] = relationship(back_populates="author")
# CRUD operations
def get_users(db: Session, skip=0, limit=20):
return db.query(User).offset(skip).limit(limit).all()
def create_user(db: Session, user: UserCreate):
db_user = User(**user.model_dump())
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_userfrom fastapi import BackgroundTasks
def send_email(to: str, subject: str):
# slow operation
time.sleep(3)
print(f"Email sent to {to}")
@app.post("/users")
async def create_user(user: UserCreate, bg: BackgroundTasks):
# ... create user
bg.add_task(send_email, user.email, "Welcome!")
return {"status": "created"}from fastapi import WebSocket
@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
await ws.accept()
try:
while True:
data = await ws.receive_text()
await ws.send_text(f"Echo: {data}")
except:
await ws.close()# uv add --dev httpx pytest
from fastapi.testclient import TestClient
from httpx import AsyncClient, ASGITransport
import pytest
# Sync testing
client = TestClient(app)
def test_read_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
def test_create_user():
response = client.post("/users", json={
"name": "Alice", "email": "a@b.com", "age": 30
})
assert response.status_code == 201
# Async testing
@pytest.mark.anyio
async def test_async():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
resp = await ac.get("/")
assert resp.status_code == 200# Dockerfile (with uv)
FROM python:3.12-slim
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev
COPY . .
CMD ["uv", "run", "fastapi", "run", "--host", "0.0.0.0"]
# Production with gunicorn
uv add gunicorn
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker