FastAPI — веб-фреймворк для создания API

FastAPI — современный асинхронный фреймворк для Python, который позволяет писать высокопроизводительные API с автоматической проверкой данных и генерацией документации.

Уровень: начальный 19 разделов ≈ 2–3 часа

1. Что такое FastAPI

FastAPI — веб-фреймворк на Python для создания REST API. Он построен на двух мощных библиотеках:

Главная особенность фреймворка — генерация документации Swagger/OpenAPI автоматически из кода: схемы запросов и ответов создаются сами.

Сравнение FastAPI и Flask

КритерийFlaskFastAPI
Асинхронностьв основном синхронныйasync/await из коробки
Валидация данныхвручную, кодомавтоматически через Pydantic
Документациянет автоматическойSwagger /docs и ReDoc /redoc
Аннотации типовне обязательныобязательная основа
Производительностьсредняявысокая
Подходит дляклассические сайты и шаблонысовременные API и микросервисы

2. Установка и первое приложение

Установи два пакета: сам фреймворк и uvicorn — ASGI-сервер для запуска приложения.

pip install fastapi uvicorn

Создаём файл main.py

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Привет, FastAPI!"}

Декоратор @app.get("/") связывает путь с функцией. Функция помечена async def, поэтому не блокирует сервер во время ожидания операций.

Запускаем приложение из терминала:

uvicorn main:app --reload

Параметр --reload автоматически перезапускает сервер при изменении кода. Это удобно во время разработки.

Формат main:app — это «имя файла: имя переменной». Если переименуешь файл или переменную, не забудь поправить команду запуска.

Открой в браузере http://localhost:8000 — увидишь JSON-ответ в виде словаря.

3. Автоматическая документация

FastAPI сам создаёт интерактивную документацию для любого приложения. Достаточно открыть два адреса:

Что можно делать в Swagger UI:

Документация строится из аннотаций типов. Чем точнее ты описываешь типы и модели, тем полезнее документация для других разработчиков.

4. Параметры пути и запроса

Параметр пути находится прямо в URL и объявляется через фигурные скобки:

from fastapi import FastAPI

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id, "name": "Иван"}

Запрос GET /users/42 вернёт {"user_id": 42, "name": "Иван"}. Аннотация int заставляет FastAPI конвертировать строку из URL в число.

Если передать в путь не число, например GET /users/abc, сервер вернёт ошибку 422 с описанием проблемы. Это происходит автоматически.

Параметры запроса (query params)

Параметры запроса идут после знака ? в URL. Они объявляются как обычные аргументы функции:

@app.get("/search")
async def search(q: str = "python", limit: int = 10):
    return {"query": q, "limit": limit}
GET /search?q=fastapi&limit=5

Значения по умолчанию делают параметры необязательными. Точные правила:

5. Тело запроса и модели Pydantic

Для сложных данных удобно отправлять JSON в теле запроса. Форму тела описывает класс, унаследованный от BaseModel:

from datetime import date

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class User(BaseModel):
    name: str
    email: str
    age: int | None = None
    birthday: date | None = None


class Item(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    price: float = Field(gt=0)
    in_stock: bool = True

Поля, отмеченные | None со значением None, являются необязательными. Обязательные поля без значения по умолчанию — name и email.

Класс Field добавляет ограничения: минимальную и максимальную длину строки, а для чисел — правила сравнения gt (больше), ge (не меньше), le (не больше).

@app.post("/users")
async def create_user(user: User):
    return {"user": user, "created": True}
Присвой функции аргумент с типом модели — и FastAPI сам прочитает тело запроса, проверит его и передаст готовый объект модели в функцию.

6. Ответы API

FastAPI сериализует возвращаемое значение в JSON автоматически. Модель Pydantic станет валидным JSON-объектом без лишних действий:

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float


@app.post("/items", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

Запись response_model=Item описывает схему ответа, а status_code=status.HTTP_201_CREATED задаёт код 201 «создано» вместо стандартного 200.

Для обработки ошибок используется исключение HTTPException:

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id != 42:
        raise HTTPException(status_code=404, detail="Товар не найден")
    return {"id": item_id, "name": "Клавиатура", "price": 4990.0}
Всегда возвращай осмысленный detail. По нему фронтенд понимает причину ошибки и показывает пользователю текст, а не общий код 500.

7. Валидация данных

Вся мощь FastAPI — в аннотациях типов. Правильно типизированная функция проверяет входные данные автоматически.

Списки и вложенные модели

class Author(BaseModel):
    name: str
    email: str


class Book(BaseModel):
    title: str
    year: int
    author: Author
    tags: list[str] = []

Модель Book содержит вложенную модель Author и список строк tags. Pydantic проверяет все уровни — каждый элемент списка и каждое поле вложенного объекта.

Query и Path с ограничениями

Объекты Query и Path задают ограничения и описание для одноимённых параметров:

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items")
async def list_items(
    q: str = Query(default="", min_length=1, max_length=50, description="Поиск по названию"),
):
    return {"query": q}


@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(ge=1, description="ID товара должен быть больше нуля"),
):
    return {"item_id": item_id}

8. Зависимости Dependency Injection

Dependency Injection (внедрение зависимостей) позволяет выносить повторяющуюся логику в отдельные функции. Типичный пример — проверка токена доступа для нескольких эндпоинтов.

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


def check_token(x_token: str = Header(...)):
    if x_token != "supersecret":
        raise HTTPException(status_code=401, detail="Доступ запрещён")
    return x_token


@app.get("/profile")
async def profile(token: str = Depends(check_token)):
    return {"user": "Иван", "role": "admin", "token": token}


@app.get("/orders")
async def orders(token: str = Depends(check_token)):
    return {"orders": [1, 2, 3]}

Что здесь происходит:

Зависимости удобно использовать для проверки прав, работы с базой данных, логирования и других повторяющихся действий.

9. Асинхронность

Функции, объявленные через async def, не блокируют сервер, пока ожидают ответа от базы данных, внешнего API или файловой системы. Пока одна операция ждёт, сервер обрабатывает другие запросы.

import httpx
from fastapi import FastAPI

app = FastAPI()


@app.get("/weather")
async def weather(city: str = "Москва"):
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={"latitude": 55.75, "longitude": 37.62},
        )
    return {"city": city, "data": response.json()}
Если функция выполняет тяжёлую синхронную работу (сложные вычисления, большой файл), не помечай её async def — иначе она заблокирует цикл событий. Для таких задач используй обычную def или очередь задач.

10. CRUD на примере заметок

Соберём всё вместе: реализуем приложение для хранения заметок. Данные будут жить в памяти процесса, в словаре.

from typing import Dict

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI()


class Note(BaseModel):
    id: int | None = None
    title: str = Field(min_length=1, max_length=100)
    text: str = ""


notes: Dict[int, Note] = {}
next_id: int = 1


@app.get("/notes")
async def get_notes():
    return list(notes.values())


@app.post("/notes", status_code=status.HTTP_201_CREATED)
async def create_note(note: Note):
    global next_id
    note.id = next_id
    notes[next_id] = note
    next_id += 1
    return note


@app.get("/notes/{note_id}")
async def get_note(note_id: int):
    note = notes.get(note_id)
    if note is None:
        raise HTTPException(status_code=404, detail="Заметка не найдена")
    return note


@app.put("/notes/{note_id}")
async def update_note(note_id: int, note: Note):
    if note_id not in notes:
        raise HTTPException(status_code=404, detail="Заметка не найдена")
    note.id = note_id
    notes[note_id] = note
    return note


@app.delete("/notes/{note_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_note(note_id: int):
    if note_id not in notes:
        raise HTTPException(status_code=404, detail="Заметка не найдена")
    del notes[note_id]

Мы реализовали четыре операции:

Словарь в памяти теряет данные при перезапуске сервера. Для постоянного хранения подключают базу данных — этим займёмся в следующем разделе.

11. База данных и запуск в продакшене

Подключение SQLAlchemy

Популярная связка: SQLAlchemy для работы с таблицами и psycopg2 как драйвер PostgreSQL.

pip install sqlalchemy psycopg2
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = "postgresql+psycopg2://user:pass@localhost/notes_db"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass

Модель таблицы и эндпоинт через зависимость get_db:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import String
from sqlalchemy.orm import Mapped, Session, mapped_column

from database import Base, SessionLocal, engine

app = FastAPI()


class Note(Base):
    __tablename__ = "notes"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))


Base.metadata.create_all(bind=engine)


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


@app.post("/notes")
async def create_note(title: str, db: Session = Depends(get_db)):
    note = Note(title=title)
    db.add(note)
    db.commit()
    db.refresh(note)
    return {"id": note.id, "title": note.title}

Сессия, созданная зависимостью, автоматически закрывается после обработки запроса.

Запуск в продакшене

Сервер uvicorn подходит и для продакшена, но при высокой нагрузке лучше запустить несколько воркеров через gunicorn:

pip install gunicorn

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

Объяснение опций:

Не забудь выключить --reload в продакшене и поставить приложение за обратный прокси (Nginx, Caddy) для раздачи статики и TLS-сертификата.

12. Тестирование FastAPI

Для тестов FastAPI использует pytest и встроенный TestClient на основе httpx. Он имитирует реальные запросы без запуска сервера:

pip install pytest httpx
# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_note():
    r = client.post("/notes", json={"title": "Купить продукты"})
    assert r.status_code == 201
    assert r.json()["title"] == "Купить продукты"

def test_note_not_found():
    r = client.get("/notes/999")
    assert r.status_code == 404

Запуск — командой pytest.

app.dependency_overrides[get_db] = override_get_db — мощный приём: тестируем эндпоинт с реальной логикой, но без внешних сервисов.
Если проект большой, выноси конфигурацию приложения в фабрику create_app() — тогда тесты создают отдельный экземпляр с тестовой базой.

13. Авторизация: OAuth2 и JWT

FastAPI предоставляет готовую схему OAuth2 с проверкой токена через зависимости. Токен JWT подписывается секретом и несёт данные пользователя:

pip install "python-jose[cryptography]" "passlib[bcrypt]"
from datetime import datetime, timedelta, timezone

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

SECRET_KEY = "замени-в-продакшене"   # из переменных окружения!
ALGORITHM = "HS256"

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_token(user_id: int) -> str:
    exp = datetime.now(timezone.utc) + timedelta(hours=1)
    return jwt.encode({"sub": str(user_id), "exp": exp}, SECRET_KEY, algorithm=ALGORITHM)

def get_current_user(token: str = Depends(oauth2_scheme)) -> int:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return int(payload["sub"])
    except JWTError:
        raise HTTPException(status_code=401, detail="Недействительный токен")

Защищённый эндпоинт:

@app.get("/profile")
async def profile(user_id: int = Depends(get_current_user)):
    return {"user_id": user_id}
Секрет JWT храни в переменной окружения, а не в коде. Время жизни токена делай коротким (час-два) и подключай refresh-токен для длинных сессий.
Пароли хэшируй через passlib: pwd_context.hash("пароль") и pwd_context.verify — bcrypt по умолчанию достаточно надёжен.

14. CORS и фоновые задачи

Для фронтенда на другом домене настрой CORS — иначе браузер заблокирует запросы:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://site.ru"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Дорогую постобработку выносят в фоновые задачи: пользователь получает ответ сразу, а отправка писем или конвертация файла выполняется после:

from fastapi import BackgroundTasks

def send_welcome(user_id: int):
    pass   # отправка письма

@app.post("/register")
async def register(name: str, background: BackgroundTasks):
    background.add_task(send_welcome, 42)
    return {"ok": True}
Фоновые задачи — не замена очереди: они выполняются в том же процессе и теряются при падении. Для надёжной обработки используй Kafka, Redis-очереди или Celery.
Middleware подключаются один раз и действуют на все эндпоинты. CORS ставь одним из первых — он влияет на все ответы приложения.

15. Загрузка файлов и потоковые ответы

Принимать файлы помогает UploadFile из fastapi:

from fastapi import File, UploadFile

@app.post("/upload")
async def upload(file: UploadFile = File(...)):
    content = await file.read()
    return {"name": file.filename, "size": len(content)}

Для больших файлов читай чанками, а не целиком:

@app.post("/upload-big")
async def upload_big(file: UploadFile = File(...)):
    size = 0
    while chunk := await file.read(1024 * 1024):
        size += len(chunk)
    return {"size": size}

Отдавать файлы потоком удобно через StreamingResponse — для видео, дампов и генерации на лету:

from fastapi.responses import StreamingResponse

@app.get("/download")
async def download():
    return StreamingResponse(
        open("big.bin", "rb"),
        media_type="application/octet-stream",
    )
Проверяй расширение и размер принимаемых файлов на сервере — иначе получишь дыру для загрузки вредоносного контента.
Для хранения файлов в облаке (S3) используй boto3: загрузка идёт потоком напрямую в бакет без занятия памяти процесса.

16. Структура приложения: роутеры и версионирование

Один файл main.py быстро разрастается до тысячи строк. FastAPI предлагает APIRouter — набор эндпоинтов, который подключается к приложению одним вызовом:

# routers/notes.py
from fastapi import APIRouter

router = APIRouter(prefix="/notes", tags=["notes"])


@router.get("")
async def list_notes():
    return [{"id": 1, "title": "Купить продукты"}]


# main.py
from routers import notes

app.include_router(notes.router)

Префикс /notes добавляется к маршрутам роутера автоматически, а tags группируют эндпоинты в /docs. Тем же механизмом делается версионирование API: роутер v1 с префиксом /api/v1 и отдельный /api/v2 позволяют менять контракт API, не ломая старых клиентов.

Разделяй файлы по областям: routers/, schemas/ (Pydantic), services/ и db/. Тогда каждый модуль читается по заголовку, а не по всему коду.
Фабрика create_app() собирает приложение из роутеров и middleware — в тестах создаётся отдельный экземпляр с тестовой базой.

17. Конфигурация: pydantic-settings и .env

Секреты и параметры сервера не должны жить в коде. Пакет pydantic-settings читает переменные окружения и файл .env в типизированный объект конфигурации:

pip install pydantic-settings

# config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Notes API"
    database_url: str = "sqlite:///notes.db"
    secret_key: str
    jwt_expire_minutes: int = 60

    model_config = {"env_file": ".env"}


settings = Settings()

Доступ к настройкам — один глобальный объект settings.database_url. Если обязательная переменная secret_key не задана, приложение падает при старте, а не в середине работы — это правильное поведение.

Файл .env добавь в .gitignore; в CI переменные подставляются через секреты раннера, а в Docker — через env_file.

Никогда не коммить .env с реальными ключами. Даже «рабочий» конфиг в репозитории — риск: одна утечка данных дороже любой экономии.

18. Миграции базы данных с Alembic

create_all() создаёт таблицы, но не меняет существующие при изменении моделей. Для эволюции схемы используется Alembic — инструмент миграций той же команды, что и SQLAlchemy:

pip install alembic
alembic init alembic
# в alembic/env.py укажи target_metadata = Base.metadata

alembic revision --autogenerate -m "add users table"
alembic upgrade head

--autogenerate сравнивает модели со схемой и генерирует скрипт миграции. Затем upgrade head применяет его, а downgrade -1 откатывает. Так изменения схемы становятся версионированными, как и код.

Порядок работы: изменил модель → сгенерировал миграцию → проверил скрипт вручную (autogenerate может ошибаться в типах) → применил → закоммитил версию.

В CI применяй миграции отдельным шагом перед тестами, а в прод — перед рестартом приложения: получишь воспроизводимый деплой без ручных SQL.
Добавление поля NOT NULL к таблице с данными — частая боль в PostgreSQL: задай server_default или сделай поле сначала nullable.

19. Типовые ошибки и чек-лист продакшена

Чаще всего на реальных проектах FastAPI спотыкаются об следующие места:

Чек-лист перед выкатом: секреты из окружения, CORS сужен до доменов, rate limiting на публичных эндпоинтах, таймауты на внешние вызовы, миграции применены, health-check /healthz, тесты в CI.

Не оставляй /docs в открытом доступе на внутренних сервисах: полная схема эндпоинтов помогает атакующему. Закрой документацию за basic-auth или VPN.
Заведи /healthz и /readyz с первой версии: оркестраторы и балансировщики будут проверять живость без сложной логики.

Что дальше

Ты научился создавать API на FastAPI: маршруты, Pydantic-модели, валидацию, зависимости, асинхронность и работу с базой данных. Следующие шаги — изучение pytest для тестирования эндпоинтов через TestClient и языка запросов SQL для сложных выборок данных.

← Вернуться к курсам