Шпаргалка по FastAPI: REST API, Pydantic, асинхронность

FastAPI — современный асинхронный Python-фреймворк для API. Автоматическая документация (Swagger/ReDoc), валидация через Pydantic, асинхронность из коробки. Производительность на уровне Node.js и Go.

Быстрый старт

pip install "fastapi[standard]"   # fastapi + uvicorn + всё необходимое

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="Описание API",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "Hello World"}

# Запуск: uvicorn main:app --reload --host 0.0.0.0 --port 8000

Документация авто-генерируется:

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

Маршрутизация

# HTTP методы
@app.get("/items")
def get_items(): return {"items": []}

@app.post("/items")
def create_item(): return {"message": "created"}

@app.put("/items/{item_id}")
def update_item(item_id: int): return {"item_id": item_id}

@app.patch("/items/{item_id}")
def partial_update(item_id: int): return {"item_id": item_id}

@app.delete("/items/{item_id}")
def delete_item(item_id: int): return {"deleted": item_id}

# Параметры пути
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}   # авто-конвертация в int, валидация

# Вложенные пути
@app.get("/users/{user_id}/posts/{post_id}")
def get_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

# Несколько методов на один путь
@app.api_route("/items", methods=["GET", "POST"])
def items_handler():
    pass

# Роутеры — разделение на модули
# routes/users.py
from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["users"])

@router.get("/")
def list_users(): return []

@router.get("/{user_id}")
def get_user(user_id: int): return {"id": user_id}

# main.py
from routes.users import router as users_router
app.include_router(users_router)

Отправка ответа

from fastapi.responses import (
    JSONResponse, HTMLResponse, PlainTextResponse,
    FileResponse, StreamingResponse, RedirectResponse
)
from fastapi import status

# JSON (по умолчанию — возврат dict/list/Pydantic-модели)
@app.get("/json")
def json_resp():
    return {"message": "OK"}              # авто-сериализация

# Явный JSONResponse (кастомные заголовки, статус)
@app.get("/custom")
def custom_resp():
    return JSONResponse(
        content={"status": "created"},
        status_code=201,
        headers={"X-Custom": "value"}
    )

# HTML
@app.get("/html")
def html_resp():
    return HTMLResponse("<h1>Привет</h1>")

# Plain text
@app.get("/text")
def text_resp():
    return PlainTextResponse("простой текст")

# Статус-коды
@app.post("/items", status_code=status.HTTP_201_CREATED)
def create(): return {"id": 1}

@app.delete("/items/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete(id: int): return None         # 204 — без тела

# Переадресация
@app.get("/old-path")
def old():
    return RedirectResponse(url="/new-path")
    # return RedirectResponse(url="/new", status_code=301)  # постоянный редирект

# Файл с сервера — скачивание
@app.get("/download/{filename}")
def download(filename: str):
    path = f"files/{filename}"
    return FileResponse(
        path=path,
        media_type="application/octet-stream",
        filename=filename                # имя в диалоге сохранения
    )

# Файл — показ в браузере (изображения, PDF)
@app.get("/image/{name}")
def image(name: str):
    return FileResponse(f"images/{name}", media_type="image/png")

# Потоковая передача (большие файлы, генерация на лету)
@app.get("/stream")
def stream():
    def generate():
        for i in range(1000):
            yield f"data: строка {i}\n\n"
    return StreamingResponse(generate(), media_type="text/plain")

# Фоновый StreamingResponse (например, CSV-экспорт)
import csv, io
@app.get("/export.csv")
def export():
    def generate_csv():
        output = io.StringIO()
        writer = csv.writer(output)
        writer.writerow(["id", "name"])
        writer.writerow([1, "Анна"])
        yield output.getvalue()
    return StreamingResponse(generate_csv(), media_type="text/csv")

Параметры запроса

from fastapi import Query
from typing import Optional

@app.get("/items/")
def get_items(
    q: Optional[str] = Query(None, min_length=3, max_length=50),
    skip: int = Query(0, ge=0, description="Пропустить N записей"),
    limit: int = Query(10, ge=1, le=100),
    sort: str = "id"
):
    return {"q": q, "skip": skip, "limit": limit}

# Параметры пути с валидацией
from fastapi import Path

@app.get("/items/{item_id}")
def get_item(
    item_id: int = Path(..., ge=1, title="Item ID"),
    q: Optional[str] = None
):
    return {"item_id": item_id, "q": q}

Получение данных запроса

from fastapi import Request, Header, Cookie

# Request — полный доступ к запросу
@app.post("/debug")
async def debug(request: Request):
    body = await request.body()          # сырые байты
    json_body = await request.json()     # распарсить JSON
    headers = dict(request.headers)      # все заголовки
    method = request.method              # GET/POST/...
    url = str(request.url)               # полный URL
    client_host = request.client.host    # IP клиента
    path_params = request.path_params    # параметры пути
    query_params = dict(request.query_params)  # ?key=value
    return {"method": method, "client": client_host}

# Отправка и получение заголовков
from fastapi import Response

@app.get("/headers")
def read_headers(
    user_agent: str | None = Header(default=None, alias="User-Agent"),
    x_token: str = Header(...)           # обязательный заголовок
):
    return {"user_agent": user_agent, "token": x_token}

@app.get("/set-headers")
def set_headers(response: Response):
    response.headers["X-Custom"] = "my-value"
    response.headers["Cache-Control"] = "no-cache"
    return {"message": "Заголовки установлены"}

# Куки — чтение и установка
@app.get("/cookies")
def read_cookies(
    session_id: str | None = Cookie(default=None),
    last_visit: str | None = Cookie(default=None)
):
    return {"session": session_id, "last_visit": last_visit}

@app.get("/set-cookie")
def set_cookie(response: Response):
    response.set_cookie(
        key="session_id",
        value="abc123",
        max_age=3600,                    # 1 час
        httponly=True,                   # недоступен JS
        secure=True,                     # только HTTPS
        samesite="lax"
    )
    return {"message": "Кука установлена"}

@app.get("/delete-cookie")
def delete_cookie(response: Response):
    response.delete_cookie("session_id")
    return {"message": "Кука удалена"}

Отправка форм

from pydantic import BaseModel, Field, validator, EmailStr
from datetime import datetime
from typing import Optional, List
from enum import Enum

# Базовая модель
class ItemBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=100, example="Ноутбук")
    description: Optional[str] = None
    price: float = Field(..., gt=0, description="Цена > 0")

# Модель для создания
class ItemCreate(ItemBase):
    category: str

# Модель для ответа (с id и датами)
class Item(ItemBase):
    id: int
    created_at: datetime

    model_config = {"from_attributes": True}  # Pydantic v2

# Модель для частичного обновления (все поля Optional)
class ItemUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = Field(None, gt=0)

# Вложенные модели
class UserBase(BaseModel):
    name: str
    email: EmailStr

class User(UserBase):
    id: int
    items: List[Item] = []

# Enum
class Category(str, Enum):
    electronics = "electronics"
    clothing = "clothing"
    food = "food"

# Валидаторы (Pydantic v2)
from pydantic import field_validator

class ItemCreate(ItemBase):
    @field_validator('title')
    @classmethod
    def title_not_empty(cls, v):
        if not v.strip():
            raise ValueError('Заголовок не может быть пустым')
        return v.strip()

CRUD с Pydantic

from fastapi import FastAPI, HTTPException, status
from typing import List

app = FastAPI()
fake_db = []
next_id = 1

@app.post("/items", response_model=Item, status_code=status.HTTP_201_CREATED)
def create(item: ItemCreate):
    global next_id
    new_item = item.model_dump()
    new_item["id"] = next_id
    new_item["created_at"] = datetime.now()
    fake_db.append(new_item)
    next_id += 1
    return new_item

@app.get("/items", response_model=List[Item])
def list_items(skip: int = 0, limit: int = 100):
    return fake_db[skip: skip + limit]

@app.get("/items/{item_id}", response_model=Item)
def get(item_id: int):
    for item in fake_db:
        if item["id"] == item_id:
            return item
    raise HTTPException(status_code=404, detail="Item not found")

@app.put("/items/{item_id}", response_model=Item)
def update(item_id: int, data: ItemCreate):
    for i, item in enumerate(fake_db):
        if item["id"] == item_id:
            updated = data.model_dump()
            updated["id"] = item_id
            updated["created_at"] = item["created_at"]
            fake_db[i] = updated
            return updated
    raise HTTPException(status_code=404, detail="Item not found")

@app.delete("/items/{item_id}")
def delete(item_id: int):
    for i, item in enumerate(fake_db):
        if item["id"] == item_id:
            del fake_db[i]
            return {"message": "Deleted"}
    raise HTTPException(status_code=404, detail="Item not found")

SQLAlchemy — работа с базой данных

pip install sqlalchemy aiosqlite
# или для PostgreSQL: pip install asyncpg psycopg2

Подключение и создание моделей

ORM-модели описывают таблицы базы данных как Python-классы. Каждый атрибут класса — колонка в таблице. SQLAlchemy берёт на себя создание таблиц (Base.metadata.create_all), генерацию SQL и преобразование строк в объекты.

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session, relationship
from datetime import datetime

# Подключение
SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Базовый класс
class Base(DeclarativeBase):
    pass

# Модели (аналог таблиц)
class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(100), nullable=False)
    description = Column(String(500), default="")
    price = Column(Float, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    owner_id = Column(Integer, ForeignKey("users.id"))
    owner = relationship("User", back_populates="items")

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    items = relationship("Item", back_populates="owner")

# Создать таблицы
Base.metadata.create_all(bind=engine)

Зависимость get_db

from fastapi import Depends

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Схемы Pydantic для SQLAlchemy

from pydantic import BaseModel, ConfigDict
from datetime import datetime

class ItemBase(BaseModel):
    title: str
    description: str | None = None
    price: float

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    created_at: datetime
    model_config = ConfigDict(from_attributes=True)   # ключевой параметр!

CRUD с реальной БД

from fastapi import FastAPI, HTTPException, status

@app.post("/items", response_model=Item, status_code=status.HTTP_201_CREATED)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)      # получить id и created_at из БД
    return db_item

@app.get("/items", response_model=list[Item])
def list_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return db.query(Item).offset(skip).limit(limit).all()

@app.get("/items/{item_id}", response_model=Item)
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/items/{item_id}", response_model=Item)
def update_item(item_id: int, data: ItemCreate, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    for key, value in data.model_dump().items():
        setattr(item, key, value)
    db.commit()
    db.refresh(item)
    return item

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    db.delete(item)
    db.commit()

# Фильтрация и поиск
@app.get("/items/search")
def search(
    q: str | None = None,
    min_price: float | None = None,
    max_price: float | None = None,
    db: Session = Depends(get_db)
):
    query = db.query(Item)
    if q:
        query = query.filter(Item.title.contains(q))
    if min_price is not None:
        query = query.filter(Item.price >= min_price)
    if max_price is not None:
        query = query.filter(Item.price <= max_price)
    return query.all()

Асинхронный SQLAlchemy (рекомендуется)

Асинхронный драйвер (aiosqlite, asyncpg) позволяет делать запросы к БД без блокировки event loop. Используйте AsyncSession вместо Session и await для всех операций. Это стандарт для production FastAPI-приложений.

pip install sqlalchemy[asyncio] aiosqlite

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

# Подключение
engine = create_async_engine("sqlite+aiosqlite:///./app.db")
async_session = async_sessionmaker(engine, expire_on_commit=False)

# Зависимость
async def get_async_db():
    async with async_session() as session:
        yield session

# Асинхронный эндпоинт
@app.get("/items", response_model=list[Item])
async def list_items(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(Item).offset(0).limit(100))
    return result.scalars().all()

@app.post("/items", response_model=Item)
async def create_item(item: ItemCreate, db: AsyncSession = Depends(get_async_db)):
    db_item = Item(**item.model_dump())
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item

Зависимости (Dependency Injection)

from fastapi import Depends

# Простая зависимость
def get_db():
    db = connect_to_db()
    try:
        yield db
    finally:
        db.close()

@app.get("/items")
def get_items(db = Depends(get_db)):
    return db.query(Item).all()

# Зависимость с параметрами
def pagination(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

@app.get("/items")
def get_items(pag = Depends(pagination)):
    return fake_db[pag["skip"]: pag["skip"] + pag["limit"]]

# Зависимость от под-зависимости
def get_token(authorization: str = Header(...)):
    return authorization.split(" ")[1]  # Bearer <token>

def get_current_user(token: str = Depends(get_token)):
    user = decode_token(token)
    if not user:
        raise HTTPException(401, "Invalid token")
    return user

@app.get("/me")
def me(user = Depends(get_current_user)):
    return user

# Глобальные зависимости (применяются ко всем маршрутам)
app = FastAPI(dependencies=[Depends(verify_api_key)])

Middleware и CORS

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],   # или ["*"] для всех
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

# Trusted hosts
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"])

# Кастомный middleware
from starlette.middleware.base import BaseHTTPMiddleware
import time

class TimerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        response.headers["X-Process-Time"] = str(duration)
        return response

app.add_middleware(TimerMiddleware)

Аутентификация

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Security

security = HTTPBearer()

@app.get("/protected")
def protected(credentials: HTTPAuthorizationCredentials = Security(security)):
    token = credentials.credentials   # сам токен
    scheme = credentials.scheme       # "Bearer"
    # ... проверить токен ...
    return {"message": "Доступ разрешён"}

# OAuth2 Password Flow
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

@app.post("/auth/login")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate(form_data.username, form_data.password)
    if not user:
        raise HTTPException(401, "Invalid credentials")
    token = create_jwt({"sub": user.id})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/me")
def me(token: str = Depends(oauth2_scheme)):
    user = decode_jwt(token)
    return user

Фоновые задачи

from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    ...  # отправка письма

@app.post("/register")
def register(email: str, background_tasks: BackgroundTasks):
    user = create_user(email)
    background_tasks.add_task(send_email, email, "Добро пожаловать!")
    return {"message": "Регистрация началась"}

# Несколько фоновых задач
background_tasks.add_task(log_activity, user.id)
background_tasks.add_task(update_stats, user.id)

Шаблоны (Jinja2)

pip install jinja2 aiofiles

Подключение шаблонов и статики

Jinja2Templates — движок шаблонов для рендеринга HTML на сервере. StaticFiles — отдача CSS, JS, изображений. Оба подключаются к приложению FastAPI и работают вместе: шаблоны ссылаются на статику через url_for('static', path='...').

from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")

Структура проекта:

project/
  main.py
  templates/
    base.html
    index.html
    items.html
  static/
    css/
    js/
    images/

Рендеринг шаблонов

templates.TemplateResponse принимает имя файла шаблона и словарь с данными (контекст). Важно: request должен быть передан в контекст всегда — Jinja2 использует его для url_for. Возвращает готовую HTML-страницу.

from fastapi import FastAPI, Request

@app.get("/")
def index(request: Request):
    return templates.TemplateResponse(
        "index.html",
        {"request": request, "title": "Главная", "items": []}
    )

@app.get("/items")
def items(request: Request):
    return templates.TemplateResponse(
        "items.html",
        {
            "request": request,           # обязательно для Jinja2Templates!
            "title": "Список товаров",
            "items": get_items(),
         }
    )

base.html — базовый шаблон

Базовый шаблон содержит общую структуру страницы (DOCTYPE, head, header, footer). Дочерние шаблоны наследуют его через {% extends "base.html" %} и переопределяют блоки ({% block content %}). Это избавляет от дублирования HTML.

<!-- templates/base.html -->
<!DOCTYPE html>
<html lang="ru">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{{ title }}</title>
    <link rel="stylesheet" href="{{ url_for('static', path='css/style.css') }}">
</head>
<body>
    <header>
        <nav>
            <a href="/">Главная</a>
            <a href="/items">Товары</a>
        </nav>
    </header>

    <main>
        {% block content %}{% endblock %}
    </main>

    <script src="{{ url_for('static', path='js/main.js') }}"></script>
</body>
</html>

Наследование и блоки

{% extends "base.html" %} — наследует всю структуру базового шаблона. {% block content %}...{% endblock %} — заполняет конкретный блок своим содержимым. Внутри блоков доступен весь контекст (переменные, циклы, условия).

<!-- templates/index.html -->
{% extends "base.html" %}

{% block content %}
    <h1>{{ title }}</h1>

    {% if items %}
        <ul>
        {% for item in items %}
            <li>
                <h3>{{ item.name }}</h3>
                <p>{{ item.price }} руб.</p>
                {% if item.description %}
                    <small>{{ item.description }}</small>
                {% endif %}
            </li>
        {% endfor %}
        </ul>
    {% else %}
        <p>Товаров пока нет</p>
    {% endif %}
{% endblock %}

Фильтры и функции

Jinja2 предоставляет встроенные фильтры (|upper, |round, |truncate) для форматирования значений прямо в шаблоне. {% if %}, {% for %}, {% macro %} — стандартные конструкции для логики и переиспользования. loop.index даёт номер итерации в цикле.

<!-- Встроенные фильтры Jinja2 -->
{{ name|upper }}
{{ price|round(2) }}
{{ text|truncate(100) }}
{{ created_at|datetimeformat('%d.%m.%Y') }}   <!-- кастомный -->
{{ content|safe }}                              <!-- без экранирования HTML -->

<!-- Условия -->
{% if user %}
    Привет, {{ user.name }}!
{% else %}
    <a href="/login">Войти</a>
{% endif %}

<!-- Цикл с индексом -->
{% for item in items %}
    <p>{{ loop.index }}. {{ item.name }}</p>    <!-- loop.index с 1, loop.index0 с 0 -->
{% endfor %}

<!-- Макросы (переиспользуемые компоненты) -->
{% macro card(item) %}
    <div class="card">
        <h3>{{ item.title }}</h3>
        <p>{{ item.price }} руб.</p>
    </div>
{% endmacro %}

{{ card(product) }}

Кастомные фильтры и глобальные функции

Вы можете расширить Jinja2 своими фильтрами (templates.env.filters) и глобальными функциями (templates.env.globals). Фильтр форматирует одно значение (my_date|datetimeformat), глобальная функция доступна везде без аргумента (current_year()).

from datetime import datetime

# Кастомный фильтр
def datetime_format(value, fmt="%d.%m.%Y"):
    if isinstance(value, str):
        value = datetime.fromisoformat(value)
    return value.strftime(fmt)

templates.env.filters["datetimeformat"] = datetime_format

# Глобальная функция (доступна во всех шаблонах)
templates.env.globals["current_year"] = lambda: datetime.now().year

<!-- Использование -->
{{ some_date|datetimeformat('%d.%m.%Y %H:%M') }}
<footer>© {{ current_year() }}</footer>

Формы и CSRF (через Jinja2)

HTML-формы отправляют данные на сервер через POST. FastAPI получает их через Form(...). Сервер валидирует данные и либо сохраняет (редирект), либо возвращает форму с ошибками. Валидация на сервере обязательна — клиентскую можно обойти.

<!-- templates/form.html -->
{% extends "base.html" %}

{% block content %}
    <h1>{{ title }}</h1>

    <form method="post" action="/items">
        <div>
            <label>Название:</label>
            <input type="text" name="name" required>
        </div>
        <div>
            <label>Цена:</label>
            <input type="number" name="price" step="0.01" required>
        </div>
        <div>
            <label>Описание:</label>
            <textarea name="description" rows="4"></textarea>
        </div>
        <button type="submit">Сохранить</button>
    </form>

    <!-- Ошибки валидации -->
    {% if errors %}
        <div class="errors">
            {% for error in errors %}
                <p class="error">{{ error }}</p>
            {% endfor %}
        </div>
    {% endif %}
{% endblock %}

# Обработка формы
from fastapi import Form

@app.post("/items")
def create_item(
    request: Request,
    name: str = Form(...),
    price: float = Form(..., gt=0),
    description: str = Form("")
):
    errors = []
    if len(name) < 3:
        errors.append("Название должно быть не менее 3 символов")
    if errors:
        return templates.TemplateResponse(
            "form.html",
            {"request": request, "title": "Новый товар", "errors": errors}
        )

    item = create_item_in_db(name=name, price=price, description=description)
    return RedirectResponse(url="/items", status_code=303)

Include — вставка фрагментов

{% include "path.html" %} вставляет содержимое другого шаблона в текущий. Используется для переиспользуемых компонентов: навигация, футер, сайдбар. В отличие от extends, include просто «вклеивает» HTML без наследования.

<!-- Переиспользуемый header -->
<!-- templates/includes/header.html -->
<header>
    <h1><a href="/">{{ site_title }}</a></h1>
    <nav>{% include "includes/nav.html" %}</nav>
</header>

<!-- Вставка в base.html -->
{% include "includes/header.html" %}

Загрузка файлов

from fastapi import UploadFile, File

@app.post("/upload")
async def upload(file: UploadFile = File(...)):
    content = await file.read()                    # весь файл в память
    # или сохранить на диск:
    with open(f"uploads/{file.filename}", "wb") as f:
        f.write(content)                           # синхронно — ОК
    return {"filename": file.filename, "size": file.size}

# Несколько файлов
@app.post("/upload-many")
async def upload_many(files: List[UploadFile] = File(...)):
    for file in files:
        await file.read()
    return {"count": len(files)}

# Параметры UploadFile
# file.filename    — имя
# file.content_type — MIME-тип
# file.size        — размер (может быть None до чтения)
# await file.read() — прочитать
# await file.seek(0) — сброс позиции

WebSocket

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Ответ: {data}")
    except WebSocketDisconnect:
        print("Клиент отключился")

# WebSocket с broadcast (чат)
class ConnectionManager:
    def __init__(self):
        self.connections: List[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.connections.append(ws)

    def disconnect(self, ws: WebSocket):
        self.connections.remove(ws)

    async def broadcast(self, message: str):
        for conn in self.connections:
            await conn.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            msg = await websocket.receive_text()
            await manager.broadcast(f"Кто-то: {msg}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Обработка ошибок

from fastapi import HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

# Исключение в маршруте
@app.get("/items/{item_id}")
def get(item_id: int):
    if item_id > 100:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "custom"}
        )
    return {"id": item_id}

# Глобальный обработчик
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    return JSONResponse(
        status_code=400,
        content={"error": str(exc)},
    )

# Кастомная обработка ошибок валидации
@app.exception_handler(RequestValidationError)
async def validation_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body},
    )

Тестирование

from fastapi.testclient import TestClient

client = TestClient(app)

def test_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post("/items", json={"title": "Test", "price": 9.99})
    assert response.status_code == 201
    assert response.json()["title"] == "Test"

def test_not_found():
    response = client.get("/items/99999")
    assert response.status_code == 404

# Переопределение зависимостей в тестах
def override_get_db():
    return test_db

app.dependency_overrides[get_db] = override_get_db
# после тестов: app.dependency_overrides.clear()

Структура проекта

project/
  main.py                  # app = FastAPI()
  models/
    item.py                # Pydantic-модели
  routes/
    items.py               # APIRouter
    users.py
  dependencies.py          # get_db, get_current_user
  services/
    items.py               # бизнес-логика
  tests/
    test_items.py
  requirements.txt

Деплой: Uvicorn и Gunicorn

Uvicorn — ASGI-сервер

Uvicorn — быстрый ASGI-сервер на uvloop и httptools. Работает как в dev-режиме с авто-перезагрузкой, так и в production.

pip install uvicorn

# Dev-режим (с авто-перезагрузкой, только для разработки!)
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# Production (без reload, несколько воркеров)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

# Аргументы командной строки
uvicorn main:app \
    --host 0.0.0.0 \           # слушать все интерфейсы
    --port 8000 \               # порт
    --workers 4 \               # число воркеров (НЕЛЬЗЯ с --reload)
    --log-level info \          # critical/error/warning/info/debug
    --access-log \              # логировать HTTP-запросы
    --proxy-headers \           # доверять X-Forwarded-* (за nginx/балансировщиком)
    --forwarded-allow-ips '*'   # разрешить Forwarded со всех IP

Gunicorn + Uvicorn workers (рекомендуется для production)

Gunicorn — проверенный WSGI-сервер. С uvicorn-воркерами становится ASGI-сервером с управлением процессами от Gunicorn (graceful restart, сигналы, логирование).

pip install gunicorn uvicorn

# Запуск
gunicorn main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --access-logfile - \        # в stdout
    --error-logfile - \
    --log-level info \
    --timeout 30 \              # таймаут воркера (сек)
    --graceful-timeout 10 \     # время на завершение запросов перед kill
    --max-requests 1000 \       # перезапуск воркера после N запросов (защита от утечек)
    --max-requests-jitter 100   # случайная добавка к max-requests

# Или через конфиг-файл gunicorn.conf.py
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
bind = "0.0.0.0:8000"
timeout = 30
graceful_timeout = 10
max_requests = 1000
max_requests_jitter = 100
accesslog = "-"
errorlog = "-"
loglevel = "info"

Что когда использовать

Systemd (Linux)

Systemd-юнит автоматически запускает приложение при старте системы и перезапускает при падении (Restart=always). Стандартный способ деплоя на голом Linux-сервере (не в Docker). Альтернатива: supervisor, docker.

# /etc/systemd/system/fastapi.service
[Unit]
Description=FastAPI app
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/opt/app
ExecStart=/opt/app/venv/bin/gunicorn main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000
Restart=always
RestartSec=5
Environment="PYTHONUNBUFFERED=1"
Environment="PATH=/opt/app/venv/bin"

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable fastapi
sudo systemctl start fastapi
sudo systemctl status fastapi

Docker

Docker упаковывает приложение со всеми зависимостями в изолированный контейнер. Гарантирует одинаковое поведение на dev-машине и сервере. Multi-stage build (отдельный слой для pip install) уменьшает размер финального образа.

# Dockerfile
FROM python:3.14-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

# Один воркер — оркестратор масштабирует контейнерами
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--proxy-headers"]

# docker-compose.yml
version: "3.8"
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/app
    depends_on:
      - db

  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: app
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

Nginx (reverse proxy)

Nginx принимает внешние запросы и проксирует их в FastAPI. Он же отдаёт статику напрямую (быстрее), терминирует HTTPS, сжимает ответы и балансирует нагрузку. WebSocket требует доп. заголовков Upgrade и Connection.

# /etc/nginx/sites-available/api
server {
    listen 80;
    server_name api.example.com;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # WebSocket
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }

    location /static/ {
        alias /opt/app/static/;
        expires 30d;
        add_header Cache-Control "public, immutable";
    }
}

Ключевые правила деплоя

• --reload никогда не использовать в production
• --proxy-headers всегда включать за reverse-прокси
• Воркеры: --workers = (2 * CPU_cores) + 1 для Gunicorn
• В Kubernetes/Docker — 1 воркер на контейнер, масштабировать подами
• max-requests помогает от утечек памяти в долгоживущих процессах
• Логи писать в stdout/stderr (12-factor app), не в файл на диске

Частые ошибки

Ресурсы

• FastAPI — официальный сайт — полная документация, туториалы, продвинутые темы
• FastAPI на GitHub — исходный код, issues, обсуждения

Что дальше

• Шпаргалка по Python: от основ до профи
• Шпаргалка по параллелизму: потоки, процессы, asyncio
• Шпаргалка по Django: модели, views, ORM, админка
• SQL шпаргалка для начинающих

Связанные шпаргалки: Python, Параллелизм, Django, SQL для начинающих