FastAPI #
1. Какая библиотека используется в FastAPI для валидации данных? #
В FastAPI для валидации данных используется Pydantic.
FastAPI использует Python type hints и Pydantic-модели для проверки входных данных, например тела запроса, query/path-параметров, а также для сериализации и валидации response models. В документации FastAPI прямо используются Pydantic-модели для описания и проверки структуры данных запроса.
Пример:
from pydantic import BaseModel
from fastapi import FastAPI
app = FastAPI()
class UserCreate(BaseModel):
username: str
email: str
age: int
@app.post("/users/")
async def create_user(user: UserCreate):
return user
Что делает Pydantic:
JSON body
↓
FastAPI передаёт данные в Pydantic-модель
↓
Pydantic проверяет типы и ограничения
↓
endpoint получает уже валидированный объект
Например, если передать:
{
"username": "alex",
"email": "alex@example.com",
"age": "not_number"
}
FastAPI вернёт ошибку валидации, потому что age должен быть int.
Важно #
В современных версиях FastAPI используется Pydantic v2, но FastAPI также долго поддерживал Pydantic v1. Сам Pydantic описывает себя как библиотеку валидации данных на основе Python type annotations.
2. Для чего используется pydantic #
Для чего используется Pydantic #
Pydantic используется для описания, проверки и преобразования данных через Python type hints.
Проще:
пришли сырые данные
↓
Pydantic проверил структуру и типы
↓
ошибочные данные отклонил
↓
валидные данные превратил в удобный Python-объект
Официально Pydantic описывает себя как библиотеку валидации данных, где валидация и сериализация управляются type annotations.
1. Валидация входных данных #
Например, пришёл JSON:
{
"username": "alex",
"age": "18"
}
Модель:
from pydantic import BaseModel
class UserCreate(BaseModel):
username: str
age: int
user = UserCreate(username="alex", age="18")
print(user)
print(type(user.age))
Результат:
username='alex' age=18
<class 'int'>
Pydantic проверил данные и привёл "18" к int.
2. Ошибки при неправильных данных #
from pydantic import BaseModel, ValidationError
class UserCreate(BaseModel):
username: str
age: int
try:
user = UserCreate(username="alex", age="not_number")
except ValidationError as error:
print(error)
Смысл:
age должен быть int
↓
"not_number" нельзя привести к int
↓
Pydantic выбрасывает ValidationError
3. Описание схемы данных #
Pydantic-модель — это контракт данных:
class UserCreate(BaseModel):
username: str
email: str
age: int
Она говорит:
username должен быть строкой
email должен быть строкой
age должен быть числом
Это удобно для API, сервисов, DTO, настроек и конфигов.
4. Использование в FastAPI #
В FastAPI Pydantic чаще всего используют для request body:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class UserCreate(BaseModel):
username: str
email: str
age: int
@app.post("/users")
async def create_user(user: UserCreate):
return user
Что происходит:
клиент отправляет JSON
↓
FastAPI передаёт JSON в Pydantic
↓
Pydantic валидирует данные
↓
endpoint получает объект UserCreate
FastAPI использует Pydantic-модели для валидации тела запроса, вложенных моделей и генерации схем для документации API.
5. Сериализация данных #
Pydantic умеет превращать модель обратно в dict или JSON:
from pydantic import BaseModel
class User(BaseModel):
id: int
username: str
user = User(id=1, username="alex")
print(user.model_dump())
print(user.model_dump_json())
Результат:
{'id': 1, 'username': 'alex'}
{"id":1,"username":"alex"}
В Pydantic v2 основной метод для преобразования модели в словарь — model_dump(), а сериализация означает преобразование модели в словарь или JSON-строку.
6. Ограничения и дополнительные проверки #
Можно задавать ограничения через Field:
from pydantic import BaseModel, Field
class UserCreate(BaseModel):
username: str = Field(min_length=3, max_length=30)
age: int = Field(ge=0, le=120)
Теперь:
username минимум 3 символа
username максимум 30 символов
age от 0 до 120
FastAPI также показывает использование Field для дополнительных валидаций и метаданных полей.
7. Настройки приложения #
Pydantic часто используют для конфигурации:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
db_url: str
redis_host: str = "localhost"
debug: bool = False
settings = Settings()
Смысл:
.env / переменные окружения
↓
Pydantic проверяет типы
↓
приложение получает объект настроек
В FastAPI-документации отдельно описано использование Pydantic-подхода для settings и environment variables.
Кратко #
Pydantic нужен для:
валидации данных
приведения типов
описания схемы данных
сериализации в dict / JSON
работы с request / response моделями в FastAPI
валидации настроек приложения
генерации JSON Schema
Главная идея:
Pydantic позволяет описывать данные обычными Python-классами
и type hints, а проверку берёт на себя.
3. Как реализован механизм внедрения зависимостей (DI | Dependency Injection)? #
В FastAPI механизм Dependency Injection реализован через функцию Depends().
Ты не создаёшь зависимости вручную внутри endpoint-а. Ты объявляешь, что endpoint зависит от какой-то функции, класса или объекта, а FastAPI сам:
читает сигнатуру endpoint-а
↓
находит параметры с Depends(...)
↓
вызывает dependency-функции
↓
получает их результат
↓
передаёт результат в endpoint
Официальная документация FastAPI описывает DI как систему, где можно указать, что path operation function зависит от чего-то ещё, что должно быть выполнено перед endpoint-ом, а FastAPI выполнит это и внедрит результат.
Простой пример #
from fastapi import Depends, FastAPI
app = FastAPI()
def get_token():
return "some-token"
@app.get("/profile")
async def profile(token: str = Depends(get_token)):
return {"token": token}
Что происходит:
GET /profile
↓
FastAPI видит Depends(get_token)
↓
вызывает get_token()
↓
получает "some-token"
↓
подставляет это значение в параметр token
↓
вызывает profile(token="some-token")
То есть Depends(get_token) говорит FastAPI:
Перед вызовом endpoint-а вызови get_token()
и результат передай сюда
Dependency — это обычный callable #
Dependency может быть:
функцией
async-функцией
классом
объектом с __call__
генератором с yield
FastAPI reference указывает, что зависимости в основном обрабатываются через Depends(), который принимает callable.
Пример с async dependency:
from fastapi import Depends, FastAPI
app = FastAPI()
async def get_current_user():
return {"id": 1, "username": "alex"}
@app.get("/me")
async def me(user: dict = Depends(get_current_user)):
return user
Dependency может сама иметь зависимости #
FastAPI строит дерево зависимостей.
from fastapi import Depends, FastAPI
app = FastAPI()
def get_db():
return "db_session"
def get_user_repository(db=Depends(get_db)):
return {"db": db}
@app.get("/users")
async def get_users(repo=Depends(get_user_repository)):
return repo
Схема:
endpoint get_users
└── get_user_repository
└── get_db
FastAPI сначала вызовет get_db(), потом передаст результат в get_user_repository(), а потом результат репозитория передаст в endpoint.
Типичный пример с БД #
from collections.abc import AsyncGenerator
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session_maker() as session:
yield session
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
db: AsyncSession = Depends(get_db),
):
result = await db.execute(...)
return result
Здесь dependency get_db() создаёт сессию БД и отдаёт её endpoint-у.
После завершения запроса код после yield/выход из context manager выполняется для очистки ресурсов. FastAPI официально поддерживает зависимости с yield для случаев, когда нужны действия после завершения dependency, например cleanup.
Как это работает внутри по шагам #
Упрощённо:
1. При регистрации route FastAPI анализирует endpoint
↓
2. Находит параметры с Depends(...)
↓
3. Строит dependency graph
↓
4. На каждый HTTP request вызывает зависимости
↓
5. Валидирует параметры зависимостей через type hints / Pydantic
↓
6. Кэширует результат dependency в рамках одного request
↓
7. Передаёт результаты в endpoint
↓
8. После ответа закрывает yield-зависимости
Пример:
async def get_current_user(
token: str = Depends(get_token),
db: AsyncSession = Depends(get_db),
):
...
FastAPI понимает:
get_current_user зависит от:
├── get_token
└── get_db
И вызывает их до get_current_user.
Кэширование dependency внутри одного request #
По умолчанию FastAPI не вызывает одну и ту же dependency несколько раз в рамках одного request, если она уже была вызвана с теми же параметрами.
Пример:
def get_db():
print("create db")
return "db"
def get_repo(db=Depends(get_db)):
return db
@app.get("/test")
async def test(
db=Depends(get_db),
repo=Depends(get_repo),
):
return {"ok": True}
get_db() обычно будет вызвана один раз за request, а результат будет переиспользован.
В Depends() есть параметр use_cache; по умолчанию он включён. Это описано в reference-документации FastAPI для dependencies.
Отключить кэш можно так:
Depends(get_db, use_cache=False)
DI в FastAPI — не совсем классический DI-container #
В некоторых фреймворках есть отдельный контейнер:
container.register(UserService)
container.resolve(UserService)
В FastAPI обычно проще:
dependency = обычная функция
Depends() = декларация зависимости
FastAPI = resolver зависимостей
То есть FastAPI DI больше завязан на HTTP request, endpoint-ы, параметры запроса, авторизацию, БД-сессии и request-scoped ресурсы.
Для чего используют DI в FastAPI #
Основные случаи:
получить DB session
получить текущего пользователя
проверить JWT/access token
проверить права доступа
получить repository/service
переиспользовать общую логику между endpoint-ами
подменять зависимости в тестах
Пример авторизации:
from fastapi import Depends, HTTPException
async def get_current_user(token: str = Depends(get_token)):
user = await decode_token(token)
if user is None:
raise HTTPException(status_code=401, detail="Unauthorized")
return user
@app.get("/me")
async def me(user=Depends(get_current_user)):
return user
Подмена зависимостей в тестах #
Один из плюсов DI — легко заменить настоящую зависимость на тестовую.
def get_db():
return real_db
def override_get_db():
return test_db
app.dependency_overrides[get_db] = override_get_db
Теперь endpoint, который использует:
db = Depends(get_db)
в тестах получит test_db, а не реальную БД.
Кратко #
DI в FastAPI реализован через Depends().
Механизм такой:
endpoint объявляет зависимости
↓
FastAPI анализирует Depends(...)
↓
строит дерево зависимостей
↓
вызывает зависимости перед endpoint-ом
↓
передаёт результаты в параметры endpoint-а
↓
после ответа закрывает yield-зависимости
Главная идея:
endpoint не создаёт зависимости сам
он только объявляет, что ему нужно
а FastAPI сам создаёт/получает это и передаёт внутрь
4. Для чего нужен Depends в FastAPI? #
Depends в FastAPI нужен, чтобы объявить зависимость для endpoint-а.
То есть endpoint говорит FastAPI:
Перед тем как вызвать меня,
сначала вызови вот эту функцию/класс,
получи результат
и передай его мне в параметр.
Официально FastAPI описывает это как систему Dependency Injection: ты объявляешь зависимости, а FastAPI сам вызывает их и передаёт результат в path operation function.
Простой пример #
from fastapi import Depends, FastAPI
app = FastAPI()
def get_token():
return "some-token"
@app.get("/profile")
async def profile(token: str = Depends(get_token)):
return {"token": token}
Что происходит:
GET /profile
↓
FastAPI видит Depends(get_token)
↓
вызывает get_token()
↓
получает "some-token"
↓
передаёт это значение в token
↓
вызывает profile(token="some-token")
Depends() принимает callable — функцию, класс или другой вызываемый объект, который FastAPI должен использовать как dependency.
Для чего реально используют Depends #
Чаще всего Depends используют для:
получения текущего пользователя
проверки JWT / access token
получения DB-сессии
проверки прав доступа
создания repository / service
переиспользования общей логики
подмены зависимостей в тестах
Пример с текущим пользователем:
from fastapi import Depends, FastAPI, HTTPException
app = FastAPI()
async def get_current_user():
user = {"id": 1, "username": "alex"}
if user is None:
raise HTTPException(status_code=401, detail="Unauthorized")
return user
@app.get("/me")
async def me(user: dict = Depends(get_current_user)):
return user
Endpoint /me сам не занимается получением пользователя. Он просто объявляет:
Мне нужен user.
Возьми его из get_current_user().
Пример с БД-сессией #
from collections.abc import AsyncGenerator
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session_maker() as session:
yield session
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
db: AsyncSession = Depends(get_db),
):
result = await db.execute(...)
return result
Здесь Depends(get_db) нужен, чтобы endpoint получил готовую AsyncSession.
Схема:
request пришёл
↓
FastAPI вызвал get_db()
↓
получил db session
↓
передал db в get_user()
↓
после завершения request закрыл session
FastAPI поддерживает зависимости с yield, чтобы можно было выполнить cleanup после завершения dependency, например закрыть соединение или сессию.
Зависимость может иметь свои зависимости #
from fastapi import Depends, FastAPI
app = FastAPI()
def get_db():
return "db"
def get_repository(db=Depends(get_db)):
return {"db": db}
@app.get("/items")
async def get_items(repo=Depends(get_repository)):
return repo
Схема:
get_items
└── get_repository
└── get_db
FastAPI сам построит цепочку зависимостей и выполнит их в правильном порядке. Документация FastAPI называет это sub-dependencies и указывает, что они могут быть вложены настолько глубоко, насколько нужно.
Depends может использоваться без передачи значения в endpoint #
Иногда зависимость нужна только для проверки, а её результат не нужен в функции.
Например, проверить токен:
from fastapi import Depends, FastAPI
app = FastAPI()
async def verify_token():
...
@app.get("/admin", dependencies=[Depends(verify_token)])
async def admin_panel():
return {"status": "ok"}
Здесь verify_token() выполнится, но его результат не будет передан в admin_panel.
FastAPI поддерживает такой вариант через dependencies=[Depends(...)] в path operation decorator. Это удобно, когда зависимость должна выполниться, но её return value не нужен внутри endpoint-а.
Главное назначение Depends:
не создавать зависимости вручную внутри endpoint-а
вынести общую логику из endpoint-ов
переиспользовать авторизацию, БД, сервисы, проверки
автоматически управлять цепочкой зависимостей
упростить тестирование через dependency overrides
По сути, Depends — это способ сказать FastAPI, какие объекты или проверки нужны endpoint-у перед выполнением.
5. Может ли Depends принимать асинхронную функцию #
Да, Depends может принимать асинхронную функцию.
Пример #
from fastapi import Depends, FastAPI
app = FastAPI()
async def get_current_user():
# например, асинхронный запрос в БД / Redis / внешний API
return {"id": 1, "username": "admin"}
@app.get("/profile")
async def get_profile(user: dict = Depends(get_current_user)):
return user
FastAPI сам вызовет dependency и дождётся результата, если это async def.
Что можно передавать в Depends
#
def sync_dependency():
return "sync"
async def async_dependency():
return "async"
Оба варианта корректны:
@app.get("/sync")
def route_1(value: str = Depends(sync_dependency)):
return value
@app.get("/async")
async def route_2(value: str = Depends(async_dependency)):
return value
FastAPI поддерживает обычные def и async def зависимости. В документации Depends() описан как механизм, принимающий callable, а раздел про зависимости показывает, что FastAPI сам решает и вызывает зависимости перед path operation.
Важный момент #
Не нужно писать так:
Depends(await get_current_user())
Это ошибка по смыслу. В Depends передаётся сама функция, а не результат её вызова:
Depends(get_current_user)
Когда делать dependency асинхронной #
async def dependency нужна, когда внутри есть асинхронные операции:
async def get_user_from_db():
user = await db.users.get(id=1)
return user
Например:
@app.get("/users/me")
async def read_me(user = Depends(get_user_from_db)):
return user
Коротко #
Depends(sync_func) — можно
Depends(async_func) — можно
Depends(func()) — обычно нельзя / неправильно
Depends(await func()) — неправильно
Для FastAPI это штатный сценарий: зависимости могут быть синхронными, асинхронными, а также зависимостями с yield; для yield документация отдельно указывает поддержку обычных и async-вариантов.
6. Как внутри устроен Depends в FastAPI? #
Внутри Depends — это не «магический вызов функции», а объект-описание зависимости, который FastAPI позже разбирает при построении и обработке route.
1. Что реально делает Depends(...)
#
Когда ты пишешь:
from fastapi import Depends
async def get_user():
return {"id": 1}
@app.get("/profile")
async def profile(user = Depends(get_user)):
return user
Depends(get_user) не вызывает get_user.
Он просто создаёт объект с настройками:
dependency = get_user
use_cache = True
scope = None
В официальной reference-документации FastAPI прямо указано: Depends принимает callable, не надо вызывать его напрямую, FastAPI вызовет его сам. В исходном коде fastapi.Depends(...) возвращает params.Depends(dependency=dependency, use_cache=use_cache, scope=scope).
То есть упрощённо:
def Depends(dependency=None, *, use_cache=True, scope=None):
return params.Depends(
dependency=dependency,
use_cache=use_cache,
scope=scope,
)
2. Где FastAPI находит Depends
#
FastAPI анализирует сигнатуру endpoint-функции.
Например:
@app.get("/items/")
async def get_items(
q: str | None = None,
user = Depends(get_user),
):
...
FastAPI смотрит параметры функции через inspect.signature(...) и для каждого параметра определяет:
q -> обычный query parameter
user -> dependency parameter
В исходниках это делает логика вокруг get_dependant(...): FastAPI получает сигнатуру callable, проходит по параметрам, анализирует каждый параметр через analyze_param(...), а если найден Depends, создаёт подзависимость через ещё один get_dependant(...).
3. FastAPI строит дерево зависимостей #
Зависимости могут зависеть от других зависимостей:
async def get_db():
return "db"
async def get_user(db = Depends(get_db)):
return {"db": db}
@app.get("/profile")
async def profile(user = Depends(get_user)):
return user
FastAPI строит примерно такое дерево:
profile
└── get_user
└── get_db
Документация FastAPI описывает это как hierarchical dependency injection system: можно определять зависимости, которые сами имеют зависимости, а FastAPI строит дерево и решает его по шагам.
4. Перед запросом FastAPI «решает» зависимости #
Когда приходит HTTP-запрос, FastAPI не сразу вызывает endpoint. Сначала он вызывает зависимости.
Упрощённая схема:
HTTP request
↓
найти route
↓
разобрать path/query/header/cookie/body параметры
↓
выполнить зависимости
↓
передать результаты зависимостей в endpoint
↓
вызвать endpoint
↓
вернуть response
Например:
async def get_user():
return {"id": 1}
@app.get("/profile")
async def profile(user = Depends(get_user)):
return user
Внутренне это похоже на:
resolved_user = await get_user()
result = await profile(user=resolved_user)
Но на практике там больше логики: валидация параметров, кэширование, sub-dependencies, yield dependencies, security scopes, request/response/background tasks.
5. Как FastAPI вызывает sync и async зависимости #
FastAPI различает тип callable:
async def dependency -> await dependency(...)
def dependency -> запустить через threadpool
yield dependency -> обработать как context manager
Документация указывает, что зависимости могут быть как async def, так и обычными def; FastAPI сам понимает, что делать.
В исходниках solve_dependencies(...) делает примерно такую развилку:
if dependency_is_generator:
solved = await solve_generator(...)
elif dependency_is_async:
solved = await call(**values)
else:
solved = await run_in_threadpool(call, **values)
То есть обычная синхронная dependency не блокирует event loop напрямую: FastAPI запускает её через threadpool. Эта логика находится в solve_dependencies(...) в fastapi/dependencies/utils.py.
6. Что такое use_cache=True
#
По умолчанию результат одной и той же dependency кэшируется в рамках одного request.
Пример:
async def get_user():
print("called")
return {"id": 1}
async def check_permissions(user = Depends(get_user)):
return True
@app.get("/profile")
async def profile(
user = Depends(get_user),
allowed = Depends(check_permissions),
):
return user
get_user может быть нужен и endpoint-у, и другой dependency. Но FastAPI по умолчанию вызовет его один раз за request и переиспользует результат.
Это поведение описано в параметре use_cache: после первого вызова dependency значение переиспользуется до конца текущего request; use_cache=False отключает это поведение.
7. Depends сам по себе ничего не делает вне FastAPI
#
Важно:
user = Depends(get_user)
Это не пользователь. Это объект зависимости.
То есть в обычной функции так не сработает:
async def service(user = Depends(get_user)):
...
Если эту функцию вызвать вручную, FastAPI не будет рядом, и user будет не результатом get_user, а объектом Depends.
Правильная идея такая:
Depends работает только там, где FastAPI сам анализирует функцию:
endpoint, dependency, router-level dependency, app-level dependency.
8. Короткая схема #
Ты пишешь:
user = Depends(get_user)
FastAPI делает:
1. Сохраняет get_user в объект Depends
2. При регистрации route анализирует сигнатуру endpoint
3. Видит параметр с Depends
4. Строит Dependant-дерево
5. При request вызывает sub-dependencies
6. Валидирует request-параметры
7. Выполняет sync/async/yield зависимости
8. Кэширует результат, если use_cache=True
9. Передаёт результат в endpoint
Главное #
Depends — это декларация зависимости, а не вызов зависимости.
user = Depends(get_user)
означает:
FastAPI, перед вызовом endpoint-а вызови get_user(),
возьми результат и подставь его в параметр user.
Именно поэтому нельзя писать:
Depends(get_user())
Потому что тогда ты передаёшь в Depends не функцию, а результат её вызова.
7. Что можно делать с Depends() в FastAPI кроме выполнения кода до endpoint? #
Главное #
Depends() нужен не только для «выполнить функцию перед endpoint». Его основная роль — описать зависимость, которую FastAPI должен сам получить, проверить, закэшировать и передать дальше. В документации FastAPI это описано как Dependency Injection-система, где зависимости могут получать параметры запроса, добавлять валидацию и попадать в OpenAPI-схему.
1. Передавать готовые данные в endpoint #
Классический сценарий — получить объект и передать его в endpoint:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
async def get_current_user():
return {"id": 1, "username": "admin"}
@app.get("/profile")
async def profile(user: Annotated[dict, Depends(get_current_user)]):
return user
То есть endpoint не сам достаёт пользователя, а получает уже готовый user.
request
↓
get_current_user()
↓
profile(user=...)
2. Выносить общие query/path/header/cookie параметры #
Например, много endpoint-ов используют skip, limit, q. Их можно вынести в dependency:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
def pagination(skip: int = 0, limit: int = 100):
return {"skip": skip, "limit": limit}
@app.get("/users")
async def get_users(params: Annotated[dict, Depends(pagination)]):
return params
@app.get("/posts")
async def get_posts(params: Annotated[dict, Depends(pagination)]):
return params
Плюс FastAPI добавит эти параметры в OpenAPI/Swagger, потому что зависимости тоже участвуют в описании параметров и валидации.
3. Делать авторизацию и проверки доступа #
Depends() часто используют для получения текущего пользователя и проверки прав:
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, status
app = FastAPI()
async def get_current_user():
return {"id": 1, "is_admin": False}
async def require_admin(
user: Annotated[dict, Depends(get_current_user)]
):
if not user["is_admin"]:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Admin access required",
)
return user
@app.delete("/users/{user_id}")
async def delete_user(
user_id: int,
admin: Annotated[dict, Depends(require_admin)],
):
return {"deleted_user_id": user_id}
Здесь require_admin — это не просто код перед endpoint, а отдельный слой доступа.
delete_user
└── require_admin
└── get_current_user
FastAPI официально поддерживает вложенные зависимости любой глубины: dependency может иметь свои sub-dependencies, а FastAPI сам решает дерево зависимостей.
4. Подключать ресурсы с lifecycle: БД, сессии, соединения #
Через dependency с yield можно выдать ресурс endpoint-у, а потом корректно закрыть его:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
async def get_db():
db = "db-session"
try:
yield db
finally:
print("close db session")
@app.get("/items")
async def get_items(db: Annotated[str, Depends(get_db)]):
return {"db": db}
Смысл:
до endpoint:
открыть ресурс
endpoint:
использовать ресурс
после endpoint:
закрыть ресурс
FastAPI поддерживает dependencies с yield; код после yield выполняется после ответа/завершения обработки, что удобно для закрытия DB-сессий и похожих ресурсов.
5. Делать dependency без возврата значения #
Иногда результат dependency не нужен. Нужно только проверить условие: токен, заголовок, API-key, доступ.
from fastapi import Depends, FastAPI, Header, HTTPException
app = FastAPI()
async def verify_token(x_token: str = Header()):
if x_token != "secret":
raise HTTPException(status_code=400, detail="Invalid token")
@app.get("/protected", dependencies=[Depends(verify_token)])
async def protected():
return {"status": "ok"}
Тут verify_token не передаётся параметром в endpoint. Она просто должна успешно выполниться. FastAPI отдельно документирует такой стиль через dependencies=[Depends(...)] в path operation decorator.
6. Вешать зависимости на router или на всё приложение #
Можно применить dependency не к одному endpoint-у, а ко всем endpoint-ам роутера или всего приложения.
from fastapi import APIRouter, Depends, FastAPI
app = FastAPI()
async def verify_auth():
return True
router = APIRouter(
prefix="/admin",
dependencies=[Depends(verify_auth)],
)
@router.get("/stats")
async def stats():
return {"stats": "secret"}
app.include_router(router)
Это удобно для общих проверок:
/admin/*
└── всегда выполнить verify_auth
FastAPI поддерживает зависимости на уровне path operation, router и всего приложения. Глобальные зависимости применяются ко всем path operations приложения.
7. Кэшировать результат dependency внутри одного request #
По умолчанию Depends(..., use_cache=True). Это значит: если одна и та же dependency нужна несколько раз в рамках одного запроса, FastAPI вызовет её один раз и переиспользует результат.
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
async def get_user():
print("called")
return {"id": 1}
async def get_permissions(
user: Annotated[dict, Depends(get_user)]
):
return ["read"]
@app.get("/me")
async def me(
user: Annotated[dict, Depends(get_user)],
permissions: Annotated[list[str], Depends(get_permissions)],
):
return {"user": user, "permissions": permissions}
get_user() нужен и endpoint-у, и get_permissions, но FastAPI может использовать кэш в рамках одного request. Параметр use_cache официально описан в reference-документации Depends.
Отключение:
Depends(get_user, use_cache=False)
8. Использовать классы как зависимости #
Dependency не обязана быть только функцией. Можно использовать класс:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class Pagination:
def __init__(self, skip: int = 0, limit: int = 100):
self.skip = skip
self.limit = limit
@app.get("/items")
async def get_items(
pagination: Annotated[Pagination, Depends(Pagination)]
):
return {
"skip": pagination.skip,
"limit": pagination.limit,
}
Это удобно, когда нужно собрать несколько параметров в один объект. В документации FastAPI есть отдельный раздел про classes as dependencies.
9. Параметризовать зависимости #
Можно сделать объект, который хранит настройки, а сам вызывается как dependency через __call__.
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
app = FastAPI()
class RoleChecker:
def __init__(self, required_role: str):
self.required_role = required_role
async def __call__(self):
user_role = "user"
if user_role != self.required_role:
raise HTTPException(status_code=403, detail="Forbidden")
return True
require_admin = RoleChecker("admin")
@app.get("/admin")
async def admin_panel(_: Annotated[bool, Depends(require_admin)]):
return {"status": "admin"}
FastAPI описывает parameterized dependencies как advanced dependency-сценарий: можно настраивать dependency без создания множества отдельных функций.
10. Менять response: headers/cookies #
Dependency может получить Response и выставить cookie или header:
from fastapi import Depends, FastAPI, Response
app = FastAPI()
def set_tracking_cookie(response: Response):
response.set_cookie(key="visited", value="true")
@app.get("/page", dependencies=[Depends(set_tracking_cookie)])
async def page():
return {"page": "home"}
FastAPI официально указывает, что Response можно объявлять в dependencies и устанавливать там cookies/headers.
11. Добавлять background tasks из dependency #
Dependency может получить BackgroundTasks и добавить фоновую задачу:
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
print(message)
def add_log_task(background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, "Request processed")
@app.post("/items", dependencies=[Depends(add_log_task)])
async def create_item():
return {"status": "created"}
BackgroundTasks работает с Dependency Injection: его можно объявлять в endpoint, dependency или sub-dependency; FastAPI объединяет задачи и запускает их после ответа.
12. Подменять зависимости в тестах #
Одна из сильных сторон Depends — dependency можно заменить в тестах:
async def get_current_user():
return real_user
async def fake_current_user():
return {"id": 1, "username": "test"}
app.dependency_overrides[get_current_user] = fake_current_user
После этого FastAPI будет вызывать fake_current_user вместо get_current_user. Это официально описано через app.dependency_overrides.
Итого #
Depends() в FastAPI используют для:
1. Получения текущего пользователя
2. Проверки авторизации и прав
3. Инъекции DB-сессии / Redis / сервисов
4. Закрытия ресурсов через yield
5. Выноса общих query/header/cookie параметров
6. Повторного использования логики между endpoint-ами
7. Вложенных зависимостей
8. Глобальных проверок на router/app уровне
9. Кэширования результата в рамках request
10. Подмены зависимостей в тестах
11. Изменения response headers/cookies
12. Добавления background tasks
Главная идея:
Depends — это способ описать, что endpoint-у нужно,
а FastAPI сам решает, как это получить, проверить,
закэшировать, передать и потом корректно завершить.
8. Как использовать Depends() для бизнес-логики в FastAPI? #
Depends() можно использовать для бизнес-логики, но лучше не складывать в него всю бизнес-операцию. Правильнее использовать его для:
1. Получения текущего пользователя
2. Проверки прав
3. Получения DB-сессии / Unit of Work
4. Создания repository/service
5. Проверки бизнес-доступа к объекту
6. Подмены зависимостей в тестах
А сама операция вроде create_order, upload_file, delete_resource обычно должна жить в service/use-case слое, который endpoint вызывает явно.
FastAPI сам решает зависимости, включая вложенные зависимости, и может подставлять их результат в endpoint. Также зависимости можно переопределять в тестах через app.dependency_overrides.
Нормальная схема #
HTTP request
↓
Depends: получить DB
↓
Depends: получить текущего пользователя
↓
Depends: собрать service
↓
endpoint: вызвать бизнес-операцию явно
↓
service: выполнить бизнес-логику
↓
response
То есть Depends() подготавливает зависимости, а не прячет весь use-case.
Пример: сервис через Depends #
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, status
from pydantic import BaseModel
app = FastAPI()
class CreateFileSchema(BaseModel):
name: str
size: int
class User:
def __init__(self, id: int, username: str, is_active: bool):
self.id = id
self.username = username
self.is_active = is_active
class FileRepository:
async def exists_by_name(self, user_id: int, name: str) -> bool:
return False
async def create(self, user_id: int, name: str, size: int) -> dict:
return {
"id": 1,
"user_id": user_id,
"name": name,
"size": size,
}
class FileService:
def __init__(self, repository: FileRepository):
self.repository = repository
async def create_file(self, user: User, data: CreateFileSchema) -> dict:
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Inactive user",
)
if data.size <= 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid file size",
)
exists = await self.repository.exists_by_name(
user_id=user.id,
name=data.name,
)
if exists:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="File already exists",
)
return await self.repository.create(
user_id=user.id,
name=data.name,
size=data.size,
)
async def get_current_user() -> User:
return User(id=1, username="admin", is_active=True)
async def get_file_repository() -> FileRepository:
return FileRepository()
async def get_file_service(
repository: Annotated[FileRepository, Depends(get_file_repository)],
) -> FileService:
return FileService(repository)
@app.post("/files")
async def create_file(
data: CreateFileSchema,
user: Annotated[User, Depends(get_current_user)],
service: Annotated[FileService, Depends(get_file_service)],
):
return await service.create_file(user=user, data=data)
Здесь:
get_current_user -> dependency
get_file_repository -> dependency
get_file_service -> dependency
create_file endpoint -> orchestration
FileService -> бизнес-логика
Это хороший вариант, потому что endpoint остаётся тонким, но бизнес-операция видна явно:
return await service.create_file(user=user, data=data)
Где именно тут бизнес-логика #
Бизнес-логика находится здесь:
async def create_file(self, user: User, data: CreateFileSchema) -> dict:
if not user.is_active:
...
if data.size <= 0:
...
exists = await self.repository.exists_by_name(...)
if exists:
...
return await self.repository.create(...)
Depends() только собирает нужные объекты.
Пример: бизнес-проверка через Depends #
Иногда проверку доступа удобно вынести в dependency.
Например: endpoint должен работать только для активного пользователя.
async def require_active_user(
user: Annotated[User, Depends(get_current_user)],
) -> User:
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Inactive user",
)
return user
Использование:
@app.post("/files")
async def create_file(
data: CreateFileSchema,
user: Annotated[User, Depends(require_active_user)],
service: Annotated[FileService, Depends(get_file_service)],
):
return await service.create_file(user=user, data=data)
Это нормально, потому что dependency отвечает за доступ, а не за создание файла.
Пример: проверка владельца объекта #
Для бизнес-доступа к конкретному объекту:
class File:
def __init__(self, id: int, owner_id: int, name: str):
self.id = id
self.owner_id = owner_id
self.name = name
class FileRepository:
async def get_by_id(self, file_id: int) -> File | None:
return File(id=file_id, owner_id=1, name="report.txt")
async def get_file_repository() -> FileRepository:
return FileRepository()
async def get_owned_file(
file_id: int,
user: Annotated[User, Depends(get_current_user)],
repository: Annotated[FileRepository, Depends(get_file_repository)],
) -> File:
file = await repository.get_by_id(file_id)
if file is None:
raise HTTPException(status_code=404, detail="File not found")
if file.owner_id != user.id:
raise HTTPException(status_code=403, detail="Forbidden")
return file
Endpoint:
@app.get("/files/{file_id}")
async def read_file(
file: Annotated[File, Depends(get_owned_file)],
):
return {
"id": file.id,
"name": file.name,
}
Здесь Depends() не просто «выполняет код до endpoint», а гарантирует endpoint-у:
сюда попадёт только файл, который существует
и принадлежит текущему пользователю
Пример: Unit of Work / DB-сессия через yield #
Для ресурсов с открытием и закрытием используют dependency с yield. FastAPI официально поддерживает dependencies с yield: код до yield отдаёт ресурс, код после yield выполняется после завершения обработки.
Упрощённый пример:
class UnitOfWork:
async def commit(self):
print("commit")
async def rollback(self):
print("rollback")
async def close(self):
print("close")
async def get_uow():
uow = UnitOfWork()
try:
yield uow
await uow.commit()
except Exception:
await uow.rollback()
raise
finally:
await uow.close()
Использование:
@app.post("/files")
async def create_file(
data: CreateFileSchema,
user: Annotated[User, Depends(require_active_user)],
service: Annotated[FileService, Depends(get_file_service)],
uow: Annotated[UnitOfWork, Depends(get_uow)],
):
result = await service.create_file(user=user, data=data)
return result
На практике uow обычно передают в repository/service, а не держат отдельно в endpoint.
Как лучше организовать файлы #
Например:
app/
├── main.py
├── api/
│ └── files.py
├── dependencies/
│ ├── auth.py
│ ├── database.py
│ └── services.py
├── services/
│ └── file_service.py
├── repositories/
│ └── file_repository.py
└── schemas/
└── file.py
Пример dependencies/services.py:
from typing import Annotated
from fastapi import Depends
from app.repositories.file_repository import FileRepository
from app.services.file_service import FileService
async def get_file_repository() -> FileRepository:
return FileRepository()
async def get_file_service(
repository: Annotated[FileRepository, Depends(get_file_repository)],
) -> FileService:
return FileService(repository)
Пример api/files.py:
from typing import Annotated
from fastapi import APIRouter, Depends
router = APIRouter(prefix="/files", tags=["files"])
@router.post("")
async def create_file(
data: CreateFileSchema,
user: Annotated[User, Depends(require_active_user)],
service: Annotated[FileService, Depends(get_file_service)],
):
return await service.create_file(user=user, data=data)
Что не стоит делать #
Плохой вариант:
async def create_file_dependency(
data: CreateFileSchema,
user: Annotated[User, Depends(get_current_user)],
service: Annotated[FileService, Depends(get_file_service)],
):
return await service.create_file(user=user, data=data)
@app.post("/files")
async def create_file(
result: Annotated[dict, Depends(create_file_dependency)],
):
return result
Почему хуже:
1. Endpoint перестаёт явно показывать бизнес-операцию
2. Основное действие спрятано внутри Depends
3. Сложнее читать flow запроса
4. Сложнее контролировать транзакции
5. Сложнее отличить подготовку контекста от выполнения команды
То есть Depends() не должен превращаться в скрытый endpoint внутри endpoint-а.
Практическое правило #
Хорошо:
Depends получает то, что нужно бизнес-операции:
- user
- db session
- repository
- service
- permissions
- owned resource
Плохо:
Depends выполняет саму бизнес-команду:
- создать заказ
- удалить файл
- списать деньги
- отправить письмо
- изменить состояние объекта
Лучше так:
@app.post("/orders")
async def create_order(
data: CreateOrderSchema,
user: Annotated[User, Depends(require_active_user)],
service: Annotated[OrderService, Depends(get_order_service)],
):
return await service.create_order(user=user, data=data)
А не так:
@app.post("/orders")
async def create_order(
order: Annotated[dict, Depends(create_order_inside_dependency)],
):
return order
Итого #
Depends() для бизнес-логики в FastAPI лучше использовать как механизм сборки контекста и зависимостей:
Depends:
- достаёт пользователя
- проверяет авторизацию
- проверяет доступ к объекту
- создаёт repository/service
- открывает и закрывает DB-сессию
- даёт endpoint-у готовые зависимости
Service:
- выполняет основную бизнес-операцию
- проверяет бизнес-правила
- меняет состояние системы
Главная формула:
Depends — подготовка и проверка контекста.
Service — бизнес-логика.
Endpoint — явный вызов use-case.
9. Почему в FastAPI используется название Depends #
Почему именно Depends
#
Название Depends логичнее понимать не как команду «выполни», а как декларацию:
этот endpoint зависит от такой-то функции / объекта / проверки
То есть когда ты пишешь:
async def profile(
user = Depends(get_current_user),
):
...
смысл такой:
profile зависит от get_current_user
FastAPI сам должен получить эту зависимость и передать результат в profile.
Официальная документация FastAPI описывает Dependency Injection так: path operation function объявляет вещи, которые требуются ей для работы, а FastAPI предоставляет эти зависимости сам. Поэтому Depends — это именно способ объявить зависимость. (
FastAPI)
Почему не Inject
#
Теоретически могли назвать так:
user = Inject(get_current_user)
Но это было бы менее точно с точки зрения кода endpoint-а.
Endpoint не говорит:
инъектируй мне пользователя
Он говорит:
я завишу от get_current_user
А уже FastAPI решает:
1. как вызвать get_current_user
2. какие sub-dependencies нужны
3. await или threadpool
4. использовать ли cache
5. что передать в endpoint
В reference-документации FastAPI прямо указано, что зависимости в основном обрабатываются специальной функцией Depends(), которая принимает callable.
Почему это не просто Before
#
Название вроде Before тоже было бы неточным.
Например:
async def get_db():
yield db
Такая dependency работает не только до endpoint, но и после endpoint — закрывает ресурс после yield. FastAPI поддерживает dependencies с yield, где код после yield выполняется после завершения обработки.
То есть Depends может описывать не просто «код до endpoint», а полноценную зависимость с жизненным циклом:
создать ресурс
↓
передать endpoint-у
↓
закрыть ресурс
Почему не Dependency
#
Можно было бы назвать:
user = Dependency(get_current_user)
Но Depends короче и лучше читается в сигнатуре функции:
user = Depends(get_current_user)
Читается почти как обычный английский:
user depends on get_current_user
То есть:
параметр user зависит от get_current_user
Главное #
Depends называется так, потому что он описывает отношение зависимости:
endpoint зависит от функции/объекта/проверки
А не просто:
выполни код перед endpoint
Более точная формула:
Depends(get_current_user)
=
FastAPI, этот параметр зависит от get_current_user.
Сам вызови, проверь, закэшируй и передай результат сюда.
10. Django vs FastAPI #
Django vs FastAPI #
Главное различие:
Django — полноценный web-фреймворк "batteries included"
FastAPI — лёгкий современный фреймворк для API
Django даёт много встроенного: ORM, миграции, admin-панель, auth, формы, шаблоны, middleware, security-механизмы. Сам Django описывает себя как high-level framework для быстрой разработки и clean pragmatic design.
FastAPI больше сфокусирован на построении API: type hints, Pydantic-валидация, автоматическая OpenAPI-документация, async-first подход и Dependency Injection. В официальной документации FastAPI описан как современный high-performance web framework для API на основе стандартных Python type hints.
Краткое сравнение #
| Критерий | Django | FastAPI |
|---|---|---|
| Основная идея | Полноценный backend-фреймворк | API-фреймворк |
| Архитектура | Более монолитная, много встроенного | Более гибкая, многое выбираешь сам |
| ORM | Встроенная Django ORM | Нет встроенной ORM |
| Админка | Есть встроенная Django Admin | Нет встроенной админки |
| API | Обычно через Django REST Framework | Встроенный основной сценарий |
| Async | Есть поддержка async views и async request stack при ASGI | Изначально хорошо ложится на async/await |
| Валидация | Forms / serializers в DRF | Pydantic / type hints |
| OpenAPI/Swagger | Обычно через DRF + доп. инструменты | Встроено из коробки |
| DI | Нет как центральной концепции | Depends() — один из ключевых механизмов |
| Лучший сценарий | Большой CRUD/backend/админка/сайт | API, микросервисы, async-сервисы |
Django имеет async-поддержку, включая async views и async request stack при ASGI, но исторически его экосистема сильнее завязана на синхронную модель. FastAPI изначально строится вокруг async def, type hints и автоматической работы с API-схемами.
Django сильнее, когда нужен полноценный backend #
Django лучше подходит, когда проекту нужны:
- пользователи
- роли и права
- админка
- ORM
- миграции
- HTML-шаблоны
- формы
- CRUD
- стандартная backend-структура
- быстрый MVP
Например:
интернет-магазин
CRM
личный кабинет
админ-панель
блог
корпоративный портал
SaaS с большим CRUD
Django хорош тем, что многое уже встроено. Не надо отдельно выбирать ORM, систему миграций, admin-панель, базовую auth-систему и часть security-инструментов. Официальный сайт Django отдельно подчёркивает rapid development, security и то, что фреймворк помогает не изобретать базовые части web-приложения заново.
FastAPI сильнее, когда нужен API-сервис #
FastAPI лучше подходит, когда проекту нужны:
- REST API
- async I/O
- микросервис
- WebSocket
- внешние интеграции
- высокая гибкость архитектуры
- строгая типизация request/response
- автогенерация Swagger/OpenAPI
Например:
notification service
auth service
payment callback service
ML API
gateway service
async API для мобильного приложения
FastAPI не заставляет использовать конкретную базу данных или ORM. В официальной документации прямо сказано, что FastAPI не требует использовать SQL database, но позволяет использовать любую БД; в примерах часто используются SQLModel/SQLAlchemy.
Django REST Framework vs FastAPI #
Если сравнивать именно API, правильнее сравнивать не просто Django, а:
Django + DRF vs FastAPI
Django REST Framework — это отдельный toolkit для построения Web API поверх Django. Он даёт serializers, authentication, permissions, browsable API и интеграцию с Django ORM.
Разница примерно такая:
Django + DRF:
- больше готовой инфраструктуры
- удобнее для CRUD вокруг Django ORM
- сильнее для проектов с админкой
- больше "магии" и слоёв
FastAPI:
- проще как API-фреймворк
- лучше читается через type hints
- удобный Depends
- проще строить чистую service/repository архитектуру
- меньше встроенного, больше решений на тебе
По скорости разработки #
Для CRUD-проекта с пользователями, админкой и моделями Django часто быстрее:
создал models
↓
миграции
↓
admin
↓
DRF serializers/viewsets
↓
готовый CRUD API
Для чистого API-сервиса FastAPI часто быстрее:
Pydantic schema
↓
endpoint
↓
Depends
↓
service
↓
response_model
↓
Swagger уже готов
По архитектуре #
В Django типичная структура:
project/
├── settings.py
├── urls.py
├── apps/
│ ├── models.py
│ ├── views.py
│ ├── serializers.py
│ ├── admin.py
│ └── migrations/
В FastAPI чаще делают более явно:
app/
├── main.py
├── api/
├── schemas/
├── services/
├── repositories/
├── dependencies/
└── db/
Django даёт больше готового каркаса. FastAPI даёт больше свободы, но архитектуру надо дисциплинированно строить самому.
По async #
FastAPI обычно удобнее, если проект изначально async:
@app.get("/users/{user_id}")
async def get_user(user_id: int):
...
Django тоже поддерживает async views:
async def my_view(request):
...
Но в Django нужно внимательнее смотреть, какие части стека реально async, а какие могут перейти в sync-режим. Официальная документация Django отдельно описывает async support и условия работы async request stack.
По ORM #
Django:
Django ORM встроена
миграции встроены
admin работает поверх моделей
FastAPI:
ORM не встроена
можно выбрать SQLAlchemy, SQLModel, Tortoise ORM, Prisma и т.д.
миграции обычно через Alembic
Это не недостаток FastAPI, а его философия: он не навязывает инфраструктуру приложения. В документации FastAPI указано, что он не требует конкретную SQL database и позволяет использовать любую БД.
Что выбрать #
Выбирай Django, если: #
- нужен большой монолит
- нужна админка
- много CRUD
- нужна встроенная ORM
- нужен backend + web-интерфейс
- проект похож на CRM / LMS / интернет-магазин / портал
- важна быстрая сборка стандартного продукта
Выбирай FastAPI, если: #
- нужен чистый REST API
- нужен микросервис
- много async I/O
- хочешь явную service/repository архитектуру
- важны type hints и Pydantic
- нужна хорошая OpenAPI-документация из коробки
- проект API-first
Практический вывод #
Django — когда нужен полноценный backend-комбайн.
FastAPI — когда нужен быстрый, гибкий и современный API-сервис.
Для твоих проектов:
Cloud storage с пользователями, MinIO, JWT, админкой
→ Django + DRF подходит хорошо
NotificationService, async Redis/PostgreSQL, микросервисный API
→ FastAPI подходит лучше
Главная разница не в том, что один «лучше», а в уровне абстракции:
Django даёт готовую систему.
FastAPI даёт удобный инструмент для сборки своей системы.
11. Каким образом FastAPI обрабатывает синхронный endpoint внутри асинхронной среды выполнения? #
Если endpoint объявлен как обычная синхронная функция через def, FastAPI не вызывает его напрямую в event loop. Он отправляет его выполнение во внешний thread pool и await-ит результат.
Официальная документация FastAPI прямо говорит: обычный def path operation запускается во внешнем threadpool и затем ожидается, потому что прямой вызов заблокировал бы сервер. То же правило применяется к синхронным dependencies.
Пример sync endpoint-а #
from fastapi import FastAPI
import time
app = FastAPI()
@app.get("/sync")
def sync_endpoint():
time.sleep(3)
return {"status": "done"}
Хотя функция обычная:
def sync_endpoint():
...
FastAPI внутри обрабатывает её примерно так:
result = await run_in_threadpool(sync_endpoint)
То есть сама функция выполняется не в event loop-потоке, а в отдельном worker thread.
Упрощённая схема #
HTTP request
↓
ASGI server: uvicorn / hypercorn
↓
FastAPI route handler
↓
FastAPI определяет: endpoint sync или async?
↓
если async def:
await endpoint(...)
↓
если def:
await run_in_threadpool(endpoint, ...)
↓
сериализация response
↓
HTTP response
В исходном коде FastAPI это видно в функции run_endpoint_function: если endpoint coroutine — вызывается await dependant.call(...); иначе используется await run_in_threadpool(dependant.call, **values).
Почему нельзя просто вызвать def напрямую
#
Потому что синхронный код может блокировать поток:
time.sleep(3)
requests.get(...)
open(...).read()
обычный sync SQL-запрос
тяжёлая обработка файла
Если выполнить это прямо внутри event loop, то на время блокировки event loop не сможет нормально обслуживать другие async-задачи.
Плохая модель:
event loop
↓
sync_endpoint()
↓
time.sleep(3)
↓
event loop заблокирован
Поэтому FastAPI делает иначе:
event loop
↓
отдать sync_endpoint в threadpool
↓
event loop свободен
↓
дождаться результата через await
Что именно используется под капотом #
FastAPI импортирует run_in_threadpool из Starlette. В исходном коде FastAPI это видно по импорту:
from starlette.concurrency import iterate_in_threadpool, run_in_threadpool
А дальше run_in_threadpool используется для синхронных endpoint-ов.
Starlette в своей документации пишет, что использует thread pool, чтобы не блокировать event loop, в том числе для synchronous endpoint-ов, созданных через def. Технически Starlette использует anyio.to_thread.run_sync для запуска синхронного кода.
Что происходит с async endpoint-ом #
@app.get("/async")
async def async_endpoint():
return {"status": "done"}
Такой endpoint вызывается напрямую через await:
result = await async_endpoint()
То есть:
async def endpoint
↓
выполняется как coroutine
↓
управление отдаётся event loop через await
А sync endpoint:
def endpoint
↓
уходит в worker thread
↓
event loop ждёт результат асинхронно
Важный момент про sync dependency #
То же самое касается Depends.
def get_user():
return {"id": 1}
@app.get("/profile")
def profile(user = Depends(get_user)):
return user
Если dependency обычная def, FastAPI тоже запускает её во внешнем threadpool. Это указано в документации FastAPI: обычные def dependencies выполняются в external threadpool.
Ограничение thread pool #
Это не бесконечное количество потоков.
Starlette указывает, что стандартный лимит thread pool — 40 tokens, то есть одновременно может выполняться ограниченное число threadpool-задач. Этот лимит общий, и FastAPI тоже использует AnyIO для sync dependencies.
AnyIO документация также описывает стандартный worker thread limiter со значением 40 для to_thread.run_sync().
То есть если у тебя много sync endpoint-ов, которые долго блокируются, может появиться очередь:
100 одновременных sync-запросов
↓
40 выполняются в threadpool
↓
остальные ждут свободный worker
Это не делает CPU-bound код реально параллельным #
Threadpool помогает не блокировать event loop, но не превращает Python-код в полноценное CPU-параллельное выполнение.
Например:
@app.get("/cpu")
def cpu_heavy():
total = 0
for i in range(100_000_000):
total += i
return {"total": total}
Такой endpoint уйдёт в threadpool, но для чистого Python CPU-bound кода всё ещё есть ограничение GIL. Event loop не будет заблокирован, но CPU может быть забит, а потоки не дадут полноценного параллелизма для Python-вычислений.
Для CPU-heavy задач обычно лучше:
- отдельный worker-процесс
- Celery / RQ / Dramatiq
- multiprocessing
- отдельный сервис
Когда def endpoint нормален
#
Нормально использовать def, если внутри синхронные библиотеки:
@app.get("/users/{user_id}")
def get_user(user_id: int):
user = sync_db_client.get_user(user_id)
return user
Например:
- sync SQLAlchemy session
- requests
- boto3
- обычная файловая операция
- legacy sync SDK
FastAPI сам отправит такой endpoint в threadpool.
Когда лучше async def
#
Лучше использовать async def, если внутри async-библиотеки:
@app.get("/users/{user_id}")
async def get_user(user_id: int):
user = await async_db_client.get_user(user_id)
return user
Например:
- asyncpg
- httpx.AsyncClient
- redis.asyncio
- aiofiles
- SQLAlchemy AsyncSession
Тогда endpoint работает напрямую в event loop и отдаёт управление на каждом await.
Нельзя делать так #
Плохой вариант:
import time
@app.get("/bad")
async def bad_endpoint():
time.sleep(3)
return {"status": "done"}
Проблема:
async def endpoint
↓
FastAPI считает его coroutine
↓
вызывает через await
↓
но внутри time.sleep(3)
↓
event loop блокируется
Если код блокирующий, лучше так:
@app.get("/better")
def better_endpoint():
time.sleep(3)
return {"status": "done"}
Или внутри async def явно вынести блокирующий кусок:
from starlette.concurrency import run_in_threadpool
@app.get("/better-async")
async def better_async_endpoint():
result = await run_in_threadpool(blocking_function)
return result
Итог #
async def endpoint
→ FastAPI вызывает через await в event loop
def endpoint
→ FastAPI запускает через Starlette run_in_threadpool
→ Starlette использует AnyIO to_thread.run_sync
→ event loop не блокируется
→ результат возвращается обратно в async-обработчик
Главная идея:
Синхронный endpoint в FastAPI работает внутри async-приложения
за счёт переноса выполнения в thread pool.
Это защищает event loop от блокировки, но не отменяет ограничений потоков, GIL и стоимости большого количества blocking-операций.
12. Как реализовать выполнение предварительной инициализации или подготовительных действий перед запуском приложения? #
Основной способ: lifespan
#
В современных FastAPI-приложениях подготовительные действия перед запуском лучше делать через lifespan.
Это механизм жизненного цикла приложения:
код до yield → выполняется перед началом приёма запросов
yield → приложение работает
код после yield → выполняется при завершении приложения
FastAPI рекомендует использовать именно lifespan для startup/shutdown-логики. Старый способ через @app.on_event("startup") считается альтернативным/deprecated; при переданном lifespan обработчики startup и shutdown уже не вызываются.
Пример #
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# Подготовительные действия перед запуском приложения
print("Init resources")
app.state.cache = {}
app.state.settings_loaded = True
yield
# Действия при остановке приложения
print("Cleanup resources")
app.state.cache.clear()
app = FastAPI(lifespan=lifespan)
@app.get("/health")
async def health():
return {"status": "ok"}
Что происходит:
uvicorn запускает приложение
↓
FastAPI вызывает lifespan
↓
выполняется код до yield
↓
приложение начинает принимать HTTP-запросы
↓
при остановке выполняется код после yield
Starlette, на которой основан FastAPI, указывает, что приложение не начнёт обслуживать входящие запросы, пока lifespan не выполнит startup-часть.
Пример с подключением Redis #
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from redis.asyncio import Redis
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = Redis(
host="localhost",
port=6379,
decode_responses=True,
)
yield
await app.state.redis.aclose()
app = FastAPI(lifespan=lifespan)
@app.get("/ping")
async def ping(request: Request):
redis: Redis = request.app.state.redis
result = await redis.ping()
return {"redis": result}
Здесь Redis-клиент создаётся один раз при старте приложения и закрывается при остановке.
Пример с HTTP-клиентом #
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI, Request
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.http_client = httpx.AsyncClient()
yield
await app.state.http_client.aclose()
app = FastAPI(lifespan=lifespan)
@app.get("/external")
async def external(request: Request):
client: httpx.AsyncClient = request.app.state.http_client
response = await client.get("https://example.com")
return {"status_code": response.status_code}
Общий ресурс удобно хранить в app.state. Starlette официально поддерживает хранение произвольного состояния на объекте приложения через app.state, а также использование lifespan state для передачи объектов в requests.
Что обычно инициализируют через lifespan
#
- подключение к Redis
- HTTP-клиент
- пул подключений к БД
- ML-модель
- кэш справочников
- проверку доступности внешних сервисов
- создание клиента S3/MinIO
- запуск внутренних async-задач
- освобождение ресурсов при shutdown
Пример FastAPI в документации показывает загрузку ML-модели до начала обработки запросов и очистку ресурсов после завершения приложения.
Старый способ: @app.on_event("startup")
#
Так тоже можно встретить:
from fastapi import FastAPI
app = FastAPI()
@app.on_event("startup")
async def startup_event():
print("Application startup")
@app.on_event("shutdown")
async def shutdown_event():
print("Application shutdown")
Но для нового кода лучше использовать lifespan.
Причина: startup и shutdown часто связаны между собой. Например, ты открыл Redis-клиент при старте и должен закрыть его при остановке. Через lifespan это лежит в одном месте:
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = Redis(...)
yield
await app.state.redis.aclose()
А через старый стиль это было бы разнесено по двум функциям:
@app.on_event("startup")
async def startup():
...
@app.on_event("shutdown")
async def shutdown():
...
FastAPI прямо указывает, что связанная startup/shutdown-логика удобнее через lifespan, поэтому этот способ сейчас рекомендуется.
Важный момент: это не Depends
#
Depends() выполняется в контексте конкретного запроса.
Depends:
на каждый request или по request-cache
lifespan:
один раз при запуске приложения
один раз при остановке приложения
То есть для подготовки приложения лучше не использовать dependency.
Плохо:
async def get_redis():
redis = Redis(...)
return redis
Так Redis может создаваться на запросах.
Лучше:
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = Redis(...)
yield
await app.state.redis.aclose()
А dependency можно использовать только для доступа к уже созданному ресурсу:
from typing import Annotated
from fastapi import Depends, Request
def get_redis(request: Request):
return request.app.state.redis
@app.get("/data")
async def data(redis: Annotated[Redis, Depends(get_redis)]):
return {"ok": await redis.ping()}
Для миграций БД #
Миграции обычно лучше не запускать внутри FastAPI lifespan.
Лучше делать отдельной командой перед стартом приложения:
alembic upgrade head
uvicorn app.main:app
Или в Docker entrypoint:
#!/bin/sh
alembic upgrade head
exec uvicorn app.main:app --host 0.0.0.0 --port 8000
Почему так лучше:
- миграции должны выполниться один раз
- при нескольких worker-ах lifespan выполнится в каждом процессе
- при рестартах контейнеров можно получить конкуренцию
- ошибки миграций проще контролировать отдельно от web-сервера
lifespan подходит для инициализации ресурсов процесса приложения, а не для одноразовых инфраструктурных операций уровня деплоя.
Важный момент про несколько workers #
Если запустить приложение так:
uvicorn app.main:app --workers 4
то у тебя будет 4 процесса приложения.
Значит startup/lifespan-код выполнится в каждом worker-процессе:
worker 1 → lifespan
worker 2 → lifespan
worker 3 → lifespan
worker 4 → lifespan
Это нормально для ресурсов процесса:
- создать Redis-клиент
- создать DB pool
- загрузить локальный кэш
Но опасно для операций, которые должны выполниться один раз глобально:
- создать администратора
- применить миграции
- пересоздать индексы
- очистить таблицы
- отправить уведомления
Такие действия лучше выносить в отдельный management command / prestart script.
Практическая схема #
from contextlib import asynccontextmanager
from typing import Annotated
import httpx
from fastapi import Depends, FastAPI, Request
from redis.asyncio import Redis
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = Redis(
host="localhost",
port=6379,
decode_responses=True,
)
app.state.http_client = httpx.AsyncClient()
yield
await app.state.http_client.aclose()
await app.state.redis.aclose()
app = FastAPI(lifespan=lifespan)
def get_redis(request: Request) -> Redis:
return request.app.state.redis
def get_http_client(request: Request) -> httpx.AsyncClient:
return request.app.state.http_client
@app.get("/health")
async def health(
redis: Annotated[Redis, Depends(get_redis)],
client: Annotated[httpx.AsyncClient, Depends(get_http_client)],
):
redis_ok = await redis.ping()
return {
"app": "ok",
"redis": redis_ok,
"http_client_closed": client.is_closed,
}
Итог #
Для подготовки приложения в FastAPI:
используй lifespan
Для доступа к подготовленным ресурсам:
используй Depends
Для одноразовых задач деплоя:
используй отдельный prestart script / management command
Главная формула:
lifespan — инициализация приложения
Depends — зависимости конкретного запроса
service — бизнес-логика
13. Как корректно организовать запуск фонового бесконечного процесса или дополнительного цикла внутри асинхронного приложения? #
Правильный способ #
В FastAPI бесконечный фоновый процесс лучше запускать через lifespan, а внутри него создавать asyncio.Task.
lifespan выполняет код до yield перед началом обработки запросов, а код после yield — при завершении приложения. FastAPI рекомендует этот механизм для startup/shutdown-логики.
Общая схема:
startup:
создать фоновую task
runtime:
приложение обрабатывает HTTP-запросы
фоновый цикл работает параллельно
shutdown:
остановить фоновую task
дождаться завершения cleanup
Базовый пример #
import asyncio
import logging
from contextlib import asynccontextmanager, suppress
from fastapi import FastAPI
logger = logging.getLogger(__name__)
async def do_background_work() -> None:
"""
Одна итерация фоновой работы:
- проверить очередь
- обновить кэш
- отправить отложенные события
- почистить старые данные
"""
logger.info("Background iteration")
await asyncio.sleep(1)
async def background_loop(stop_event: asyncio.Event) -> None:
try:
while not stop_event.is_set():
try:
await do_background_work()
except Exception:
logger.exception("Background loop iteration failed")
try:
await asyncio.wait_for(stop_event.wait(), timeout=10)
except asyncio.TimeoutError:
pass
except asyncio.CancelledError:
logger.info("Background loop cancelled")
raise
finally:
logger.info("Background loop cleanup finished")
@asynccontextmanager
async def lifespan(app: FastAPI):
stop_event = asyncio.Event()
task = asyncio.create_task(
background_loop(stop_event),
name="background-loop",
)
app.state.background_task = task
try:
yield
finally:
stop_event.set()
task.cancel()
with suppress(asyncio.CancelledError):
await task
app = FastAPI(lifespan=lifespan)
@app.get("/health")
async def health():
return {"status": "ok"}
asyncio.create_task() планирует корутину для конкурентного выполнения, а документация Python отдельно подчёркивает, что ссылку на созданную task нужно сохранять, иначе задача может быть потеряна сборщиком мусора.
Почему не просто while True в startup
#
Плохо:
@asynccontextmanager
async def lifespan(app: FastAPI):
while True:
await do_background_work()
yield
Проблема:
lifespan не дойдёт до yield
↓
приложение не начнёт принимать запросы
Starlette, на которой основан FastAPI, прямо указывает: приложение не начинает обслуживать входящие запросы, пока startup-часть lifespan не завершилась.
Поэтому бесконечный цикл нужно не выполнять напрямую, а запускать как отдельную задачу:
task = asyncio.create_task(background_loop(stop_event))
Почему не BackgroundTasks
#
BackgroundTasks в FastAPI предназначен для задач, которые нужно выполнить после отправки конкретного HTTP-ответа. Например: отправить email, записать лог, обработать файл после запроса.
То есть это не механизм для постоянного фонового цикла.
Плохо:
@app.post("/start")
async def start(background_tasks: BackgroundTasks):
background_tasks.add_task(background_loop)
return {"status": "started"}
Почему плохо:
1. BackgroundTasks привязан к request/response
2. Он не является lifecycle-механизмом приложения
3. Сложнее контролировать shutdown
4. Можно случайно запустить несколько одинаковых циклов
5. Нет нормальной модели владения задачей
Для постоянного процесса:
lifespan + asyncio.create_task
Для фоновой задачи после конкретного запроса:
BackgroundTasks
Как должен быть устроен бесконечный цикл #
Хороший фоновый цикл должен:
1. Иметь await внутри цикла
2. Уметь завершаться
3. Обрабатывать CancelledError
4. Не глушить CancelledError полностью
5. Логировать ошибки итераций
6. Не падать навсегда из-за одной ошибки
7. Корректно освобождать ресурсы в finally
Пример нормального цикла:
async def background_loop(stop_event: asyncio.Event) -> None:
try:
while not stop_event.is_set():
try:
await do_background_work()
except Exception:
logger.exception("Background iteration failed")
await asyncio.sleep(10)
except asyncio.CancelledError:
logger.info("Background task was cancelled")
raise
finally:
logger.info("Cleanup background resources")
Важно: CancelledError лучше не проглатывать. Его нужно либо не ловить вообще, либо поймать для cleanup и снова сделать raise.
Вариант для consumer-а очереди #
Например, есть asyncio.Queue, и отдельный процесс должен постоянно читать задачи:
import asyncio
import logging
from contextlib import asynccontextmanager, suppress
from fastapi import FastAPI, Request
logger = logging.getLogger(__name__)
async def queue_worker(queue: asyncio.Queue) -> None:
try:
while True:
item = await queue.get()
try:
logger.info("Processing item: %s", item)
await asyncio.sleep(1)
except Exception:
logger.exception("Failed to process item")
finally:
queue.task_done()
except asyncio.CancelledError:
logger.info("Queue worker cancelled")
raise
@asynccontextmanager
async def lifespan(app: FastAPI):
queue: asyncio.Queue = asyncio.Queue()
task = asyncio.create_task(queue_worker(queue), name="queue-worker")
app.state.queue = queue
app.state.queue_worker_task = task
try:
yield
finally:
task.cancel()
with suppress(asyncio.CancelledError):
await task
app = FastAPI(lifespan=lifespan)
@app.post("/tasks")
async def add_task(request: Request):
await request.app.state.queue.put({"type": "demo"})
return {"status": "queued"}
Здесь:
HTTP endpoint кладёт задачу в queue
↓
queue_worker обрабатывает её независимо
↓
при shutdown worker отменяется
Несколько фоновых процессов #
Можно запустить несколько задач:
@asynccontextmanager
async def lifespan(app: FastAPI):
tasks = [
asyncio.create_task(background_loop_1(), name="loop-1"),
asyncio.create_task(background_loop_2(), name="loop-2"),
asyncio.create_task(background_loop_3(), name="loop-3"),
]
app.state.background_tasks = tasks
try:
yield
finally:
for task in tasks:
task.cancel()
for task in tasks:
with suppress(asyncio.CancelledError):
await task
Но если задач становится много, лучше смотреть в сторону anyio.create_task_group(). Starlette в документации по lifespan отдельно рекомендует рассматривать anyio.create_task_group() для управления асинхронными задачами.
AnyIO task group имеет собственный cancel scope, и всю группу можно отменить через tg.cancel_scope.cancel().
Важный момент про несколько workers #
Если приложение запущено так:
uvicorn app.main:app --workers 4
то будет 4 процесса.
Значит lifespan выполнится 4 раза:
worker 1 → background_loop
worker 2 → background_loop
worker 3 → background_loop
worker 4 → background_loop
Это нормально для локальных задач каждого процесса:
- локальный health-check
- локальный in-memory cache refresh
- обслуживание локального состояния worker-а
Но опасно для глобальных задач:
- списание денег
- отправка email-рассылки
- обработка общей очереди без lock-а
- периодическое создание отчётов
- изменение общих данных в БД
Для глобального фонового процесса лучше использовать отдельный worker:
FastAPI process → принимает HTTP
Worker process → выполняет фоновые задачи
Redis/RabbitMQ → очередь между ними
FastAPI в документации по BackgroundTasks также указывает, что для тяжёлых фоновых вычислений или работы в нескольких процессах/серверах могут быть полезны инструменты вроде Celery с брокером сообщений.
Что нельзя делать #
1. Нельзя запускать бесконечный цикл без await
#
async def bad_loop():
while True:
do_sync_work()
Такой цикл забьёт event loop.
Нужно:
async def good_loop():
while True:
await do_async_work()
await asyncio.sleep(10)
2. Нельзя создавать task и не хранить ссылку #
Плохо:
asyncio.create_task(background_loop())
Лучше:
task = asyncio.create_task(background_loop())
app.state.background_task = task
Python-документация прямо указывает сохранять ссылку на task для надёжного fire-and-forget сценария.
3. Нельзя использовать BackgroundTasks для вечного процесса
#
Плохо:
background_tasks.add_task(infinite_loop)
BackgroundTasks — для задач после конкретного ответа, а не для жизненного цикла приложения.
4. Нельзя забывать про shutdown #
Плохо:
@asynccontextmanager
async def lifespan(app: FastAPI):
asyncio.create_task(background_loop())
yield
Здесь задача запускается, но не управляется.
Лучше:
task = asyncio.create_task(background_loop())
try:
yield
finally:
task.cancel()
with suppress(asyncio.CancelledError):
await task
Практическое правило #
lifespan
запуск и остановка долгоживущих фоновых процессов
asyncio.create_task
запуск корутины параллельно с приложением
task.cancel() + await task
корректная остановка
BackgroundTasks
короткая задача после конкретного HTTP-ответа
Celery / RQ / Dramatiq
тяжёлые, надёжные или распределённые фоновые задачи
Итоговая формула:
Для бесконечного фонового цикла в FastAPI:
запускай task в lifespan
храни ссылку на task
делай await внутри цикла
обрабатывай ошибки
отменяй task на shutdown
не используй BackgroundTasks для вечных процессов
14. Почему FastAPI не поддерживает XML сериализацию из коробки? #
FastAPI не поддерживает XML-сериализацию «из коробки», потому что его основной стек построен вокруг:
Python type hints
↓
Pydantic-модели
↓
JSON-compatible data
↓
JSONResponse
↓
OpenAPI/Swagger
По умолчанию FastAPI возвращает JSON-ответы. В документации прямо указано: если вернуть dict, list, Pydantic-модель и т.п., FastAPI сериализует это в JSON; при необходимости можно вернуть Response напрямую и указать свой media_type.
Почему JSON поддержан лучше #
FastAPI создавался как фреймворк для современных API, где основной формат обмена — JSON. Сам FastAPI описывает себя как web framework для построения API на основе стандартных Python type hints.
Для JSON у FastAPI есть естественная цепочка:
class UserOut(BaseModel):
id: int
username: str
@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
return {"id": user_id, "username": "admin"}
FastAPI понимает:
какая схема данных
как валидировать
как сериализовать
как описать это в OpenAPI
как показать это в Swagger
С XML так просто не получается.
Почему XML сложнее сериализовать автоматически #
У JSON структура ближе к Python-объектам:
dict -> object
list -> array
str -> string
int -> number
bool -> boolean
None -> null
А у XML есть дополнительные особенности:
<user id="1">
<username>admin</username>
</user>
Тут сразу появляются вопросы:
id — это атрибут или отдельный тег?
username — это тег или текст?
как хранить namespaces?
важен ли порядок элементов?
как обрабатывать mixed content?
как валидировать через XSD?
как маппить XML в Pydantic-модель?
То есть универсального очевидного правила:
Pydantic model -> XML
нет.
Для JSON такое правило естественное. Для XML нужно выбирать конкретную стратегию сериализации.
Почему это не ограничение ASGI/FastAPI как сервера #
FastAPI может вернуть XML. Просто он не делает автоматическую XML-сериализацию моделей по умолчанию.
Пример:
from fastapi import FastAPI, Response
app = FastAPI()
@app.get("/users/{user_id}")
async def get_user(user_id: int):
xml = f"""
<user>
<id>{user_id}</id>
<username>admin</username>
</user>
"""
return Response(
content=xml,
media_type="application/xml",
)
То есть проблема не в том, что FastAPI «не умеет отправить XML». Он умеет отправить любой текстовый/байтовый ответ через Response. В документации FastAPI описано, что можно вернуть Response напрямую и самостоятельно управлять содержимым, статусом, headers и media type.
XML request тоже можно принять, но вручную #
Например:
from fastapi import FastAPI, Request
import xml.etree.ElementTree as ET
app = FastAPI()
@app.post("/users")
async def create_user(request: Request):
raw_body = await request.body()
root = ET.fromstring(raw_body)
user_id = int(root.findtext("id"))
username = root.findtext("username")
return {
"id": user_id,
"username": username,
}
FastAPI позволяет получить сырой Request напрямую, без автоматической валидации тела через Pydantic. Это нужно как раз для нестандартных форматов данных.
Почему FastAPI не делает это сам #
Основные причины:
1. FastAPI ориентирован на JSON API
2. Pydantic естественно работает с Python-объектами и JSON-подобными структурами
3. OpenAPI/Swagger-интеграция проще и стандартнее для JSON
4. XML имеет неоднозначный mapping в объектные модели
5. Для XML часто нужны отдельные правила: attributes, namespaces, XSD, порядок тегов
6. XML чаще нужен для legacy/enterprise-интеграций, а не как основной формат новых API
OpenAPI сам по себе поддерживает разные media types, включая JSON и XML, но media type нужно явно описывать в request/response. FastAPI же по умолчанию строит удобный путь именно для JSON.
Как правильно добавить XML в FastAPI #
Для response:
from fastapi import Response
@app.get(
"/users/{user_id}",
responses={
200: {
"content": {
"application/xml": {
"example": """
<user>
<id>1</id>
<username>admin</username>
</user>
"""
}
},
"description": "XML response",
}
},
)
async def get_user_xml(user_id: int):
xml = f"""
<user>
<id>{user_id}</id>
<username>admin</username>
</user>
"""
return Response(
content=xml,
media_type="application/xml",
)
Для request:
from fastapi import Request
import xml.etree.ElementTree as ET
@app.post("/users/xml")
async def create_user_from_xml(request: Request):
raw_body = await request.body()
root = ET.fromstring(raw_body)
return {
"id": int(root.findtext("id")),
"username": root.findtext("username"),
}
Итог #
FastAPI не поддерживает XML-сериализацию из коробки не потому, что это технически невозможно, а потому что его стандартная модель такая:
type hints + Pydantic + JSON + OpenAPI
XML требует отдельной стратегии:
как строить теги
как работать с атрибутами
как учитывать namespaces
как валидировать XML
как описывать это в документации
Поэтому в FastAPI XML обычно делают явно:
Response(..., media_type="application/xml")
+
ручный парсинг request.body()
+
при необходимости отдельная XML-библиотека
15. Как FastAPI обрабатывает входящий запрос? #
Общая схема #
FastAPI обрабатывает входящий HTTP-запрос примерно так:
client
↓
ASGI server: Uvicorn / Hypercorn
↓
ASGI-приложение FastAPI
↓
middleware
↓
routing
↓
парсинг path/query/header/cookie/body
↓
Depends / security / validation
↓
endpoint
↓
response_model / сериализация
↓
middleware после ответа
↓
ASGI send
↓
client
FastAPI работает как ASGI-приложение. ASGI-приложение вызывается как async callable с тремя аргументами: scope, receive, send; Uvicorn использует именно этот ASGI-интерфейс для связи с приложением.
1. Запрос сначала принимает ASGI-сервер #
Например, ты запускаешь:
uvicorn app.main:app
Uvicorn принимает HTTP-запрос от клиента и передаёт его FastAPI-приложению через ASGI-интерфейс:
async def app(scope, receive, send):
...
Грубо:
scope -> информация о соединении: path, method, headers, client, scheme
receive -> канал для получения тела запроса
send -> канал для отправки ответа
ASGI-документация описывает scope как словарь с данными соединения, receive как async callable для входящих событий, а send как async callable для исходящих событий.
2. Запрос проходит через middleware #
До попадания в конкретный endpoint запрос проходит через middleware:
@app.middleware("http")
async def add_header(request, call_next):
response = await call_next(request)
response.headers["X-App"] = "FastAPI"
return response
Middleware может выполнить код:
до endpoint-а
после endpoint-а
FastAPI-документация описывает middleware как функцию, которая работает с каждым request до конкретной path operation и с каждым response перед возвратом клиенту.
3. Starlette/FastAPI ищет подходящий route #
FastAPI построен поверх Starlette, поэтому routing-слой во многом опирается на Starlette.
Например:
@app.get("/users/{user_id}")
async def get_user(user_id: int):
return {"user_id": user_id}
При запросе:
/users/123
routing должен определить:
path -> /users/{user_id}
method -> GET
params -> {"user_id": "123"}
Starlette routing поддерживает path-параметры и конвертеры вроде int, float, uuid, path; endpoint может быть обычной или async-функцией.
4. FastAPI создаёт объект Request
#
После того как запрос попал в route handler, FastAPI работает уже не с голыми scope, receive, send, а с объектом Request.
Упрощённо:
request = Request(scope, receive, send)
В исходном коде FastAPI видно, что route-обёртка создаёт Request(scope, receive, send), использует AsyncExitStack для request-scoped cleanup и затем отправляет сформированный Response обратно через ASGI.
5. FastAPI читает тело запроса #
Если endpoint ожидает body:
from pydantic import BaseModel
class UserIn(BaseModel):
username: str
age: int
@app.post("/users")
async def create_user(data: UserIn):
return data
FastAPI должен:
1. Прочитать body
2. Понять content-type
3. Если это JSON — распарсить JSON
4. Передать данные в Pydantic-валидацию
5. При ошибке вернуть 422
В исходном коде FastAPI get_request_handler читает body, отдельно обрабатывает form-data, JSON content type, JSON decode error и ошибку парсинга тела запроса.
6. FastAPI решает зависимости Depends
#
Перед вызовом endpoint-а FastAPI решает дерево зависимостей.
Пример:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
async def get_current_user():
return {"id": 1}
@app.get("/profile")
async def profile(
user: Annotated[dict, Depends(get_current_user)],
):
return user
Перед вызовом profile() FastAPI вызовет get_current_user() и подставит результат в параметр user.
Официальная документация FastAPI описывает Dependency Injection как механизм, через который path operation function объявляет то, что ей нужно для работы, а FastAPI сам предоставляет эти зависимости.
В исходном коде route handler вызывает solve_dependencies(...), передаёт туда request, dependant, тело запроса и AsyncExitStack; после этого получает solved_result.values, которые потом попадут в endpoint.
7. FastAPI валидирует параметры #
FastAPI собирает значения из разных частей запроса:
path params
query params
headers
cookies
body
dependencies
Например:
@app.get("/items/{item_id}")
async def get_item(
item_id: int,
limit: int = 10,
):
return {
"item_id": item_id,
"limit": limit,
}
Запрос:
/items/abc?limit=test
не пройдёт валидацию, потому что item_id и limit ожидаются как int.
То есть endpoint обычно вызывается уже с подготовленными Python-значениями:
get_item(item_id=123, limit=10)
а не с сырыми строками из URL.
8. FastAPI вызывает endpoint #
После парсинга, зависимостей и валидации FastAPI вызывает endpoint.
Для async def:
@app.get("/async")
async def async_endpoint():
return {"ok": True}
вызов примерно такой:
result = await async_endpoint()
Для обычного def:
@app.get("/sync")
def sync_endpoint():
return {"ok": True}
FastAPI отправляет выполнение в threadpool:
result = await run_in_threadpool(sync_endpoint)
В исходном коде FastAPI функция run_endpoint_function проверяет, является ли endpoint coroutine; если да — делает await dependant.call(**values), если нет — вызывает endpoint через run_in_threadpool.
9. Endpoint возвращает результат #
Endpoint может вернуть разные типы:
return {"id": 1}
или:
return UserOut(id=1, username="admin")
или готовый response:
from fastapi import Response
return Response(
content="<ok>true</ok>",
media_type="application/xml",
)
Если endpoint возвращает обычные Python-данные, FastAPI превращает их в HTTP-response. Если возвращается готовый Response, FastAPI использует его напрямую.
Starlette response-классы отвечают за отправку ASGI-сообщений через send; Response содержит content, status_code, headers, media_type и может быть вызван как ASGI-приложение.
10. FastAPI применяет response_model
#
Если указан response_model, FastAPI валидирует и фильтрует выходные данные:
from pydantic import BaseModel
class UserOut(BaseModel):
id: int
username: str
@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
return {
"id": user_id,
"username": "admin",
"password": "secret",
}
Клиент получит:
{
"id": 1,
"username": "admin"
}
password будет отфильтрован, потому что его нет в UserOut.
FastAPI-документация указывает, что response_model используется для документации, валидации, конвертации и фильтрации выходных данных.
11. Response проходит обратно через middleware #
После endpoint-а и сериализации response возвращается обратно через middleware-цепочку.
endpoint
↓
response_model / сериализация
↓
middleware after-response код
↓
ASGI send
↓
client
Это позволяет middleware добавить headers, посчитать время обработки, залогировать response и т.д. FastAPI-документация прямо описывает, что middleware может обработать response перед возвратом клиенту.
12. Выполняется cleanup #
После обработки запроса FastAPI должен корректно завершить request-scoped ресурсы:
async def get_db():
db = Session()
try:
yield db
finally:
await db.close()
Для этого FastAPI использует AsyncExitStack. В исходном коде FastAPI видно, что при обработке request создаются request/function stacks и кладутся в scope; это нужно в том числе для dependency cleanup.
Сжатая схема внутри FastAPI #
1. ASGI server вызывает FastAPI(scope, receive, send)
2. Middleware получает Request
3. Router ищет подходящий path + method
4. FastAPI создаёт route handler
5. Читается body:
- JSON
- form-data
- bytes
- либо body отсутствует
6. solve_dependencies():
- path params
- query params
- headers
- cookies
- body
- Depends
- security dependencies
7. Если есть ошибки:
- RequestValidationError
- обычно HTTP 422
8. Если ошибок нет:
- async endpoint -> await endpoint(...)
- sync endpoint -> run_in_threadpool(endpoint, ...)
9. Результат endpoint-а:
- если Response -> вернуть как есть
- если dict/list/model -> сериализовать
10. response_model:
- валидировать
- преобразовать
- отфильтровать поля
11. Response отправляется через ASGI send
12. Закрываются yield-dependencies и request-scoped ресурсы
Мини-пример полного прохода #
from typing import Annotated
from fastapi import Depends, FastAPI
from pydantic import BaseModel
app = FastAPI()
class UserIn(BaseModel):
username: str
class UserOut(BaseModel):
id: int
username: str
async def get_current_user():
return {"id": 10, "role": "admin"}
@app.post("/users/{user_id}", response_model=UserOut)
async def update_user(
user_id: int,
data: UserIn,
current_user: Annotated[dict, Depends(get_current_user)],
):
return {
"id": user_id,
"username": data.username,
"password": "hidden",
"updated_by": current_user["id"],
}
Запрос:
POST /users/5
Content-Type: application/json
{
"username": "alex"
}
Что делает FastAPI:
1. Находит route POST /users/{user_id}
2. Достаёт user_id из path
3. Конвертирует user_id в int
4. Читает JSON body
5. Валидирует body через UserIn
6. Выполняет get_current_user через Depends
7. Вызывает update_user(user_id=5, data=UserIn(...), current_user=...)
8. Получает dict
9. Применяет response_model=UserOut
10. Убирает password и updated_by
11. Возвращает JSON
Ответ:
{
"id": 5,
"username": "alex"
}
Главное #
FastAPI — это слой над ASGI/Starlette, который добавляет:
- type hints based validation
- Pydantic-схемы
- Dependency Injection
- автоматическую сериализацию
- response_model-фильтрацию
- OpenAPI-документацию
То есть FastAPI не просто вызывает endpoint. Он сначала строит и решает весь контекст запроса, а endpoint получает уже подготовленные и провалидированные значения.
16. Какой web-server встроен в FastAPI? #
В сам FastAPI не встроен полноценный web-server в смысле сетевого сервера, который сам слушает сокеты.
FastAPI — это ASGI web framework, то есть приложение, которое должно запускаться через ASGI-сервер. Обычно для этого используется Uvicorn. В документации FastAPI сказано, что для запуска FastAPI-приложения нужен ASGI server program вроде Uvicorn; именно Uvicorn используется по умолчанию командой fastapi run.
Правильная схема #
client
↓
Uvicorn / Hypercorn / другой ASGI server
↓
FastAPI application
↓
routes / Depends / validation / response
То есть:
Uvicorn — web-server / ASGI-server
FastAPI — web-framework / ASGI-application
Uvicorn официально описывает себя как ASGI web server implementation for Python.
Почему часто говорят «FastAPI использует Uvicorn» #
Потому что обычно приложение запускают так:
uvicorn app.main:app --reload
или через новую команду:
fastapi run app/main.py
Но технически это не значит, что Uvicorn «встроен внутрь FastAPI как часть фреймворка». Более точная формулировка:
FastAPI-приложение запускается ASGI-сервером.
Чаще всего этим сервером является Uvicorn.
Можно ли использовать не Uvicorn #
Да. FastAPI может запускаться и через другие ASGI-серверы, например Hypercorn. В документации FastAPI указано, что есть несколько альтернатив ASGI-серверов.
Итог #
Встроенный web-server в FastAPI:
как отдельный сервер — нет
Стандартный ASGI-сервер для запуска:
Uvicorn
Что делает FastAPI:
routing, validation, Depends, OpenAPI, response handling
Что делает Uvicorn:
принимает HTTP-запросы по сети
управляет event loop
передаёт запросы FastAPI через ASGI
17. Что такое BackgroundTasks в FastAPI? #
Что такое BackgroundTasks
#
BackgroundTasks в FastAPI — это механизм для запуска функции после отправки HTTP-ответа клиенту.
То есть клиент не ждёт завершения этой операции.
Пример сценариев:
- записать лог
- отправить email
- удалить временный файл
- обновить простую статистику
- выполнить небольшую пост-обработку после request
FastAPI описывает BackgroundTasks как способ определить задачи, которые должны выполниться после возврата ответа. Это полезно для операций, из-за которых клиент не должен ждать завершения обработки.
Простой пример #
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_log(message: str) -> None:
with open("log.txt", mode="a", encoding="utf-8") as file:
file.write(message + "\n")
@app.post("/send")
async def send_message(
background_tasks: BackgroundTasks,
):
background_tasks.add_task(
write_log,
"Message was sent",
)
return {"status": "accepted"}
Что произойдёт:
1. Клиент отправляет POST /send
2. Endpoint добавляет задачу через background_tasks.add_task(...)
3. FastAPI сразу возвращает response
4. После отправки response выполняется write_log(...)
Как это работает логически #
request
↓
endpoint
↓
background_tasks.add_task(...)
↓
return response
↓
response отправлен клиенту
↓
BackgroundTasks выполняет добавленные задачи
Starlette, на которой основан FastAPI, описывает background task как задачу, прикреплённую к response, которая запускается только после отправки response.
Можно передавать аргументы #
def send_email(email: str, text: str) -> None:
print(f"Send email to {email}: {text}")
@app.post("/register")
async def register_user(
background_tasks: BackgroundTasks,
):
background_tasks.add_task(
send_email,
"user@example.com",
"Welcome",
)
return {"status": "registered"}
add_task() принимает:
1. функцию
2. positional args
3. keyword args
Можно использовать async-функцию #
async def async_write_log(message: str) -> None:
await some_async_logger.write(message)
@app.post("/event")
async def create_event(
background_tasks: BackgroundTasks,
):
background_tasks.add_task(
async_write_log,
"event created",
)
return {"status": "ok"}
Можно добавлять и обычные def, и async def.
Синхронный код Starlette запускает через thread pool, чтобы не блокировать event loop. Это касается endpoint-функций и background tasks.
BackgroundTasks можно использовать в dependency
#
Например, dependency добавляет логирование:
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_audit_log(action: str) -> None:
print(f"Audit: {action}")
def add_audit_log(
background_tasks: BackgroundTasks,
):
background_tasks.add_task(
write_audit_log,
"endpoint was called",
)
@app.post("/items", dependencies=[Depends(add_audit_log)])
async def create_item():
return {"status": "created"}
FastAPI умеет объединять BackgroundTasks, объявленные в endpoint, dependency и sub-dependency, в один общий объект.
Чем отличается от asyncio.create_task
#
BackgroundTasks привязан к конкретному HTTP-запросу и response.
BackgroundTasks:
выполнить задачу после ответа на конкретный request
asyncio.create_task:
запустить корутину независимо внутри event loop
Для задачи после ответа:
@app.post("/notify")
async def notify(background_tasks: BackgroundTasks):
background_tasks.add_task(send_email, "user@example.com")
return {"status": "queued"}
Для долгоживущего процесса лучше использовать lifespan + asyncio.create_task, а не BackgroundTasks.
Чем отличается от Celery / RQ / Dramatiq #
BackgroundTasks выполняется внутри того же процесса FastAPI.
FastAPI process
├── HTTP endpoint
└── BackgroundTasks
Это значит:
- нет отдельной очереди
- нет гарантированного retry
- задача может пропасть при падении процесса
- задача не распределяется между отдельными worker-сервисами
- не подходит для тяжёлых и критичных операций
Для тяжёлых или надёжных фоновых задач лучше использовать отдельную очередь задач. В документации FastAPI отдельно указано, что для тяжёлых фоновых вычислений или работы в нескольких процессах/серверах можно использовать более мощные инструменты вроде Celery.
Когда использовать BackgroundTasks
#
Подходит:
- короткие некритичные задачи
- логирование
- отправка простого уведомления
- очистка временного файла
- действия после успешного ответа
Не подходит:
- бесконечный цикл
- периодический scheduler
- тяжёлая CPU-bound обработка
- критичная бизнес-операция
- обработка платежей
- задачи, которые нельзя потерять
- долгие задачи на минуты/часы
Важный пример: не делать так #
Плохо:
@app.post("/order")
async def create_order(
background_tasks: BackgroundTasks,
):
background_tasks.add_task(save_order_to_db)
return {"status": "created"}
Почему плохо:
Клиент уже получил "created",
но заказ ещё может не сохраниться.
Критичную бизнес-логику надо выполнить до ответа:
@app.post("/order")
async def create_order():
order = await order_service.create_order()
return order
А в BackgroundTasks можно положить побочные действия:
@app.post("/order")
async def create_order(
background_tasks: BackgroundTasks,
):
order = await order_service.create_order()
background_tasks.add_task(
send_order_email,
order.id,
)
return order
Главное #
BackgroundTasks — это не отдельный worker и не очередь задач.
Это механизм FastAPI/Starlette для выполнения небольших функций
после того, как response уже отправлен клиенту.
Формула:
Нужно сделать что-то после конкретного request
и потеря этой операции не критична
→ BackgroundTasks
Нужен вечный цикл
→ lifespan + asyncio.create_task
Нужна надёжная очередь, retry, отдельные worker-ы
→ Celery / RQ / Dramatiq
18. Какой вид Response по умолчанию используется в FastAPI? #
По умолчанию: JSONResponse
#
В FastAPI по умолчанию используется JSON-ответ — обычно это JSONResponse.
То есть если endpoint возвращает обычный dict, list, Pydantic-модель и т.п., FastAPI превращает это в JSON и отправляет клиенту с media type:
Content-Type: application/json
Документация FastAPI прямо указывает: по умолчанию FastAPI возвращает JSON responses; если response_model не задан, FastAPI использует jsonable_encoder и помещает результат в JSONResponse.
Пример #
from fastapi import FastAPI
app = FastAPI()
@app.get("/user")
async def get_user():
return {
"id": 1,
"username": "admin",
}
Фактически FastAPI сделает примерно так:
JSONResponse(
content={
"id": 1,
"username": "admin",
}
)
Клиент получит:
{
"id": 1,
"username": "admin"
}
Если указан response_model
#
from pydantic import BaseModel
class UserOut(BaseModel):
id: int
username: str
@app.get("/user", response_model=UserOut)
async def get_user():
return {
"id": 1,
"username": "admin",
"password": "secret",
}
FastAPI сначала применит response_model, отвалидирует и отфильтрует данные, а затем вернёт JSON-ответ. response_model используется для документации, валидации, конвертации и фильтрации выходных данных.
Ответ:
{
"id": 1,
"username": "admin"
}
Можно изменить response class #
Например, на конкретном endpoint-е:
from fastapi.responses import HTMLResponse
@app.get("/page", response_class=HTMLResponse)
async def page():
return "<h1>Hello</h1>"
Или глобально для приложения:
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
app = FastAPI(
default_response_class=ORJSONResponse,
)
FastAPI позволяет задавать default_response_class на уровне приложения или роутера.
Важное уточнение #
Если ты возвращаешь готовый Response, FastAPI уже не оборачивает его в JSONResponse:
from fastapi import Response
@app.get("/xml")
async def xml():
return Response(
content="<status>ok</status>",
media_type="application/xml",
)
Здесь ответ будет XML, а не JSON.
Итог #
По умолчанию FastAPI использует JSONResponse.
dict/list/Pydantic model
↓
jsonable_encoder / response_model
↓
JSONResponse
↓
application/json
Но response можно заменить:
response_class=HTMLResponse
response_class=PlainTextResponse
response_class=FileResponse
response_class=StreamingResponse
default_response_class=ORJSONResponse
return Response(...)
19. Какие ORM используют с FastAPI? | Использовали ли вы SQLAlchemy с FastAPI? #
Какие ORM используют с FastAPI #
FastAPI не навязывает ORM. Можно использовать любую БД и любой слой доступа к данным. В официальной документации FastAPI прямо указано, что фреймворк не требует SQL-базу и не привязывает к конкретной ORM; в примерах сейчас часто используется SQLModel.
Чаще всего с FastAPI используют:
1. SQLAlchemy
2. SQLModel
3. Tortoise ORM
4. Peewee
5. Ormar / Piccolo / GINO — реже
6. Без ORM: asyncpg, psycopg, raw SQL
1. SQLAlchemy #
Самый распространённый и взрослый вариант.
FastAPI + SQLAlchemy + Alembic
Обычно используют:
- SQLAlchemy ORM
- SQLAlchemy Core
- Alembic для миграций
- asyncpg для PostgreSQL async-подключения
SQLAlchemy официально поддерживает asyncio для Core и ORM через asyncio-compatible dialects.
Пример стека:
FastAPI
↓
Depends(get_db)
↓
SQLAlchemy AsyncSession
↓
Repository / Service
↓
PostgreSQL
Пример dependency:
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/app_db"
engine = create_async_engine(DATABASE_URL)
async_session_maker = async_sessionmaker(
engine,
expire_on_commit=False,
)
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session_maker() as session:
yield session
Использование:
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
db: Annotated[AsyncSession, Depends(get_db)],
):
user = await db.get(User, user_id)
return user
2. SQLModel #
SQLModel — это библиотека от автора FastAPI, построенная поверх SQLAlchemy и Pydantic. В документации FastAPI сказано, что SQLModel построен на SQLAlchemy и Pydantic и был сделан как удобное сочетание для FastAPI-приложений с SQL-БД.
SQLModel = SQLAlchemy + Pydantic-подход
Плюсы:
- удобно для небольших и средних проектов
- меньше дублирования между ORM-моделью и Pydantic-схемой
- хорошо ложится на FastAPI
Минусы:
- меньше гибкости, чем чистый SQLAlchemy
- для сложных проектов часто всё равно переходят к полноценному SQLAlchemy
3. Tortoise ORM #
Tortoise ORM — async ORM, по стилю похожая на Django ORM.
FastAPI + Tortoise ORM + Aerich
Tortoise имеет официальную интеграцию с FastAPI: регистрирует ORM с setup/teardown внутри lifespan приложения.
Плюсы:
- изначально async
- простой синтаксис
- похожа на Django ORM
- удобно для CRUD
Минусы:
- экосистема меньше, чем у SQLAlchemy
- для сложных SQL-запросов SQLAlchemy обычно сильнее
Примерно так выглядит использование:
user = await User.get(id=user_id)
await User.create(username="admin")
4. Raw SQL / asyncpg без ORM #
Иногда ORM вообще не используют.
FastAPI + asyncpg
FastAPI + psycopg
FastAPI + databases
Такой подход подходит, когда:
- нужны сложные SQL-запросы
- важна максимальная прозрачность
- проект небольшой
- ORM даёт больше сложности, чем пользы
Минус — больше ручного кода:
- вручную писать SQL
- вручную маппить строки БД в DTO/схемы
- вручную следить за транзакциями
Что чаще выбрать #
SQLAlchemy — основной production-вариант
SQLModel — удобно для простых FastAPI CRUD/API
Tortoise — удобно, если хочется Django-like async ORM
Raw SQL — когда нужен полный контроль над SQL
Для серьёзного backend-проекта я бы чаще выбирал:
FastAPI + SQLAlchemy 2.x + Alembic + PostgreSQL
Итог #
FastAPI не имеет встроенной ORM.
На практике чаще всего используют:
- SQLAlchemy — основной зрелый вариант
- SQLModel — удобный вариант под FastAPI
- Tortoise ORM — async Django-like вариант
- raw SQL / asyncpg — когда нужен полный контроль