FastAPI — веб-фреймворк для создания API
FastAPI — современный асинхронный фреймворк для Python, который позволяет писать высокопроизводительные API с автоматической проверкой данных и генерацией документации.
1. Что такое FastAPI
FastAPI — веб-фреймворк на Python для создания REST API. Он построен на двух мощных библиотеках:
- Starlette — легковесный ASGI-каркас, отвечающий за обработку HTTP-запросов.
- Pydantic — библиотека валидации данных по аннотациям типов.
Главная особенность фреймворка — генерация документации Swagger/OpenAPI автоматически из кода: схемы запросов и ответов создаются сами.
Сравнение FastAPI и Flask
| Критерий | Flask | FastAPI |
|---|---|---|
| Асинхронность | в основном синхронный | 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 сам создаёт интерактивную документацию для любого приложения. Достаточно открыть два адреса:
/docs— Swagger UI, удобная панель для тестирования./redoc— ReDoc, документация в стиле читабельного справочника.
Что можно делать в Swagger UI:
- видеть все эндпоинты приложения и их методы;
- нажимать Try it out и отправлять реальные запросы;
- указывать значения параметров и тела запроса;
- видеть ответ сервера с кодом статуса и телом ответа;
- просматривать схемы всех моделей Pydantic.
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Значения по умолчанию делают параметры необязательными. Точные правила:
- аргумент без значения по умолчанию — обязательный query-параметр;
- аргумент со значением по умолчанию — необязательный;
- тип
intилиfloat— автоматическая проверка числа; - тип
bool— принимаетtrue,1,yesи т. п.
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}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}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]}Что здесь происходит:
- функция
check_token— зависимость, она выполняется перед эндпоинтом; - если токен не подходит — поднимается ошибка 401 и эндпоинт не выполняется;
- возвращённое значение зависимости передаётся в функцию как аргумент;
- логика проверки описана один раз и переиспользуется в любом количестве эндпоинтов.
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]Мы реализовали четыре операции:
GET /notes— получить все заметки;POST /notes— создать заметку с новым id;PUT /notes/{id}— полностью обновить заметку;DELETE /notes/{id}— удалить заметку.
11. База данных и запуск в продакшене
Подключение SQLAlchemy
Популярная связка: SQLAlchemy для работы с таблицами и psycopg2 как драйвер PostgreSQL.
pip install sqlalchemy psycopg2from 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Объяснение опций:
-w 4— четыре рабочих процесса;-k uvicorn.workers.UvicornWorker— воркер для ASGI-приложений;-b 0.0.0.0:8000— слушать все сетевые интерфейсы на порту 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подменяй реальные зависимости заглушками; - в CI прогоняй тесты после линтера, см. GitHub Actions.
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}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}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",
)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, не ломая старых клиентов.
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.
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 может ошибаться в типах) → применил → закоммитил версию.
server_default или сделай поле сначала nullable.19. Типовые ошибки и чек-лист продакшена
Чаще всего на реальных проектах FastAPI спотыкаются об следующие места:
- синхронная CPU-работа внутри async-эндпоинта блокирует весь сервер — выноси её в фоновые задачи или очередь;
- отсутствие пагинации: список, растущий без лимита, убивает и базу, и клиента;
- response_model без схемы: наружу утекают поля ORM-моделей — хэши, служебные атрибуты;
- UI не обрабатывает 422: «быстрая» проверка FastAPI должна отражаться в формах фронтенда;
- секрет по умолчанию или в коде — всегда бери его из Settings;
- nullable-поля без ограничений пропускают мусор в базу.
Чек-лист перед выкатом: секреты из окружения, CORS сужен до доменов, rate limiting на публичных эндпоинтах, таймауты на внешние вызовы, миграции применены, health-check /healthz, тесты в CI.
/healthz и /readyz с первой версии: оркестраторы и балансировщики будут проверять живость без сложной логики.Что дальше
Ты научился создавать API на FastAPI: маршруты, Pydantic-модели, валидацию, зависимости, асинхронность и работу с базой данных. Следующие шаги — изучение pytest для тестирования эндпоинтов через TestClient и языка запросов SQL для сложных выборок данных.