Как обрабатывается request в FastAPI?

Жизненный цикл request в FastAPI

FastAPI построена на Starlette и использует асинхронный ASGI сервер. Понимание этого процесса критично для отладки и оптимизации.

1. Полный путь request-response

Клиент (HTTP запрос)
    ↓
ASGI сервер (Uvicorn)
    ↓
FastAPI приложение
    ↓
Марутизация (Router matching)
    ↓
Dependencies (внедрение зависимостей)
    ↓
Валидация (Pydantic)
    ↓
Path parameters, Query parameters, Body
    ↓
Handler function (обработчик маршрута)
    ↓
Response (JSON сериализация)
    ↓
Клиент (HTTP ответ)

2. Маршрутизация (Routing)

FastAPI проверяет маршруты в порядке их определения:

from fastapi import FastAPI

app = FastAPI()

# Порядок ВАЖЕН!
@app.get("/users/me")  # Этот маршрут проверяется первым
async def get_current_user():
    return {"user": "current"}

@app.get("/users/{user_id}")  # Этот проверяется вторым
async def get_user(user_id: int):
    return {"user_id": user_id}

# Если переставить порядок, /users/me будет обработан как /users/{user_id}
# где user_id="me", и возникнет ошибка валидации

Процесс маршрутизации:

from fastapi import FastAPI
from fastapi.routing import APIRoute

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

# FastAPI парсит маршрут
# Шаблон: /items/{item_id}
# Path параметр: item_id (тип: int)
# Query параметры: q (тип: str, опциональный)

# При запросе GET /items/42?q=hello
# item_id = 42
# q = "hello"

3. Обработка зависимостей (Dependencies)

FastAPI имеет встроенную систему внедрения зависимостей:

from fastapi import FastAPI, Depends, Header, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthCredentials

app = FastAPI()

# Зависимость 1: токен из заголовка
def verify_token(authorization: str = Header(None)):
    if not authorization:
        raise HTTPException(status_code=401, detail="Token required")
    return authorization

# Зависимость 2: пользователь из БД
def get_current_user(token: str = Depends(verify_token)):
    # Проверить токен и получить пользователя
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"user_id": 1, "username": "john"}

# Использование зависимостей
@app.get("/items/")
async def read_items(current_user = Depends(get_current_user)):
    # FastAPI автоматически:
    # 1. Вызовет verify_token(authorization="Bearer token")
    # 2. Затем вызовет get_current_user(token="Bearer token")
    # 3. Результат передаст в read_items
    return {"items": [], "user": current_user}

# Порядок выполнения:
# verify_token() → get_current_user() → read_items()

Граф зависимостей:

from fastapi import FastAPI, Depends

def dependency_a():
    print("A")
    return "a"

def dependency_b(a = Depends(dependency_a)):
    print("B")
    return f"b({a})"

def dependency_c(a = Depends(dependency_a), b = Depends(dependency_b)):
    print("C")
    return f"c({a}, {b})"

@app.get("/")
async def main(c = Depends(dependency_c)):
    print("Main")
    return {"result": c}

# Порядок выполнения:
# A
# B
# A (кэшируется в рамках одного request)
# C
# Main

4. Валидация параметров (Pydantic)

FastAPI использует Pydantic для автоматической валидации:

from fastapi import FastAPI, Query, Path, Body
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)  # price > 0
    description: Optional[str] = None
    tax: float = Field(default=0, ge=0, le=100)

@app.post("/items/")
async def create_item(item: Item):  # Body валидируется Pydantic
    # Если JSON невалидный, FastAPI вернёт 422 Unprocessable Entity
    # с описанием ошибок валидации
    return item

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., gt=0),  # Path параметр, должен быть > 0
    q: str = Query(None, max_length=50)  # Query параметр
):
    return {"item_id": item_id, "q": q}

# При запросе GET /items/abc?q=test
# Ошибка: "item_id must be a valid integer"

# При запросе POST /items/ с {"name": "", "price": -10}
# Ошибка: name требует min_length=1, price должен быть > 0

5. Обработка request body

from fastapi import FastAPI, Form, File, UploadFile
from typing import List

app = FastAPI()

# JSON body
@app.post("/items/")
async def create_item(item: dict):
    return item

# Form data
@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

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

# Multiple files
@app.post("/upload-multiple/")
async def upload_files(files: List[UploadFile]):
    return {"count": len(files)}

# Mixed: JSON body + form data + files
@app.post("/mixed/")
async def mixed(item: dict = Body(...), file: UploadFile = File(...)):
    return {"item": item, "file": file.filename}

6. Middleware и Hooks

FastAPI позволяет перехватывать request-response:

from fastapi import FastAPI, Request
from fastapi.middleware import Middleware
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# Middleware 1: Логирование времени выполнения
@app.middleware("http")
async def log_request_time(request: Request, call_next):
    start_time = time.time()
    
    # Передать request дальше по цепи
    response = await call_next(request)
    
    # После обработки добавить заголовок
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

# Middleware 2: CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Порядок выполнения middleware (как матрёшка):
# 1. Request → log_request_time
# 2. Request → CORSMiddleware
# 3. Request → Router → Handler
# 4. Response → CORSMiddleware
# 5. Response → log_request_time → Клиент

7. Exception handling

FastAPI автоматически обрабатывает исключения:

from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_error_handler
)
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

# Встроенная обработка
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id < 0:
        raise HTTPException(
            status_code=400,
            detail="Item ID must be positive"
        )
    if item_id > 1000:
        raise HTTPException(
            status_code=404,
            detail="Item not found"
        )
    return {"item_id": item_id}

# Кастомная обработка исключений
class CustomException(Exception):
    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message

@app.exception_handler(CustomException)
async def custom_exception_handler(request, exc):
    return JSONResponse(
        status_code=400,
        content={"error": exc.code, "message": exc.message}
    )

# Использование
@app.get("/custom/")
async def custom_endpoint():
    raise CustomException(
        code="CUSTOM_ERROR",
        message="Something went wrong"
    )

8. Асинхронность в FastAPI

from fastapi import FastAPI
import asyncio

app = FastAPI()

# Асинхронный handler (рекомендуется)
@app.get("/async/")
async def async_handler():
    # Может использовать await
    await asyncio.sleep(1)  # Не блокирует другие requests
    return {"status": "ok"}

# Синхронный handler (блокирует)
@app.get("/sync/")
def sync_handler():
    # FastAPI автоматически запустит в thread pool
    time.sleep(1)  # Блокирует, но в отдельном потоке
    return {"status": "ok"}

# FastAPI автоматически выбирает способ выполнения

9. Response модели

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class UserBase(BaseModel):
    username: str
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool

    class Config:
        from_attributes = True

# Response модель определяет, какие поля возвращать
@app.post("/users/", response_model=User)
async def create_user(user: UserCreate) -> User:
    # Функция может возвращать UserCreate,
    # но FastAPI вернёт только поля из User
    return {"id": 1, "username": user.username, "email": user.email, "is_active": True}

# Клиент получит:
# {"id": 1, "username": "...", "email": "...", "is_active": true}
# password НЕ будет в ответе (хотя был в handler)

10. Полный пример обработки request

from fastapi import FastAPI, Depends, Header, HTTPException, Request
from pydantic import BaseModel
import time

app = FastAPI()

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

# 1. Зависимость: проверка токена
def verify_token(authorization: str = Header(None)):
    if not authorization or authorization != "Bearer valid-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return authorization

# 2. Зависимость: получить пользователя
def get_current_user(token: str = Depends(verify_token)):
    return {"user_id": 1, "username": "john"}

# 3. Middleware: логирование
@app.middleware("http")
async def log_middleware(request: Request, call_next):
    print(f"Request: {request.method} {request.url.path}")
    start = time.time()
    response = await call_next(request)
    print(f"Time: {time.time() - start:.3f}s")
    return response

# 4. Handler
@app.post("/items/")
async def create_item(
    item: Item,  # Валидация Pydantic
    current_user = Depends(get_current_user)  # Внедрение зависимостей
):
    # Обработка
    return {
        "item": item,
        "created_by": current_user["username"]
    }

# Порядок выполнения при POST /items/:
# 1. Uvicorn получает HTTP запрос
# 2. log_middleware: логирование request
# 3. Router: находит /items/ маршрут
# 4. verify_token: проверяет заголовок Authorization
# 5. get_current_user: получает пользователя
# 6. Pydantic: валидирует JSON body в Item
# 7. create_item: выполняет обработчик
# 8. JSON сериализация ответа
# 9. log_middleware: добавляет время выполнения
# 10. Response отправляется клиенту

Оптимизация обработки request

# 1. Используй async для I/O операций
@app.get("/")
async def fast_io():
    result = await some_async_db_query()
    return result

# 2. Кэшируй результаты зависимостей
def expensive_dependency():
    # Выполняется один раз за request
    return expensive_operation()

# 3. Используй response_model для сериализации
@app.get("/", response_model=UserResponse)
async def get_user():
    # FastAPI оптимизирует сериализацию
    pass

# 4. Сжимай ответы
@app.get("/", responses={200: {"content": {"application/json": {...}}}})
async def compressed():
    pass

Итог

Request в FastAPI проходит:

1. Маршрутизация → поиск подходящего маршрута
2. Зависимости → внедрение зависимостей (граф)
3. Валидация → Pydantic проверяет типы
4. Handler → выполнение функции
5. Сериализация → преобразование в JSON
6. Response → возврат клиенту

Всё это асинхронно и автоматизировано для производительности.