Веб-API на FastAPI и Pydantic

# main.py
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI(title="Todo API", version="1.0")

class TaskIn(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    done: bool = False

class TaskOut(TaskIn):
    id: int

tasks: dict[int, TaskOut] = {}

@app.get("/tasks", response_model=list[TaskOut])
def list_tasks(done: bool | None = None, limit: int = 20, offset: int = 0):     # query-параметры
    items = [t for t in tasks.values() if done is None or t.done == done]
    return items[offset: offset + limit]

@app.get("/tasks/{task_id}", response_model=TaskOut)
def get_task(task_id: int):                                                      # параметр пути
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail="Задача не найдена")
    return tasks[task_id]

@app.post("/tasks", response_model=TaskOut, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskIn):                                                   # тело JSON
    new = TaskOut(id=len(tasks) + 1, **task.model_dump())
    tasks[new.id] = new
    return new

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int):
    if tasks.pop(task_id, None) is None:
        raise HTTPException(404, "Задача не найдена")

pip install "fastapi[standard]"
fastapi dev main.py                  # или: uvicorn main:app --reload
# http://127.0.0.1:8000/docs — интерактивная документация Swagger UI
curl -X POST localhost:8000/tasks -H "Content-Type: application/json" -d '{"title": "Выучить FastAPI"}'

from typing import Annotated
from fastapi import Depends, Header, HTTPException
from sqlalchemy.orm import Session

def get_session():
    with SessionLocal() as session:
        yield session                                  # после ответа сессия закроется

SessionDep = Annotated[Session, Depends(get_session)]

def get_current_user(session: SessionDep, authorization: str = Header()):
    token = authorization.removeprefix("Bearer ")
    user = session.scalars(select(User).where(User.token == token)).one_or_none()
    if user is None:
        raise HTTPException(401, "Неверный токен", headers={"WWW-Authenticate": "Bearer"})
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

class Pagination(BaseModel):
    limit: int = Field(20, le=100)
    offset: int = 0

@app.get("/me/orders", response_model=list[OrderOut])
def my_orders(user: CurrentUser, session: SessionDep, page: Annotated[Pagination, Depends()]):
    stmt = select(Order).where(Order.user_id == user.id).limit(page.limit).offset(page.offset)
    return session.scalars(stmt).all()

app/
├── main.py              # создание FastAPI(), подключение роутеров, middleware
├── config.py            # настройки (pydantic-settings)
├── db.py                # engine, SessionLocal, get_session
├── models/              # модели SQLAlchemy
├── schemas/             # модели Pydantic: UserCreate, UserOut…
├── api/
│   ├── deps.py          # общие зависимости: текущий пользователь, пагинация
│   └── routes/
│       ├── users.py     # router = APIRouter(prefix="/users", tags=["users"])
│       └── orders.py
├── services/            # бизнес-логика, не зависит от HTTP
└── tests/

# app/main.py
app = FastAPI()
app.include_router(users.router, prefix="/api/v1")
app.include_router(orders.router, prefix="/api/v1")

from fastapi import BackgroundTasks, Request
from fastapi.middleware.cors import CORSMiddleware
import time

@app.post("/signup")
def signup(user: UserCreate, background: BackgroundTasks):
    created = create_user(user)
    background.add_task(send_welcome_email, created.email)   # после ответа; для тяжёлого — Celery/RQ/arq
    return created

@app.middleware("http")
async def timing(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    response.headers["X-Process-Time"] = f"{time.perf_counter() - start:.4f}"
    return response

app.add_middleware(CORSMiddleware, allow_origins=["https://my-frontend.example"], allow_methods=["*"])

@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(status_code=409, content={"error": {"code": exc.code, "message": str(exc)}})

from fastapi.testclient import TestClient
from app.main import app, get_session

def fake_session():
    yield TestingSession()

app.dependency_overrides[get_session] = fake_session
client = TestClient(app)

def test_create_and_get_task():
    r = client.post("/tasks", json={"title": "Тест"})
    assert r.status_code == 201
    task_id = r.json()["id"]
    assert client.get(f"/tasks/{task_id}").json()["title"] == "Тест"

def test_validation_error():
    r = client.post("/tasks", json={"title": ""})
    assert r.status_code == 422
    assert r.json()["detail"][0]["loc"] == ["body", "title"]