REST API: проектирование, документация и лучшие практики

REST: не просто "HTTP + JSON"

REST (Representational State Transfer) — архитектурный стиль, описанный Роем Филдингом в 2000 году. Большинство "REST API" в реальности — это RPC поверх HTTP (CRUD по URL). Настоящий REST имеет шесть ограничений, из которых на практике применяют три-четыре.

Но это не важно. Важно проектировать API, которым приятно пользоваться: предсказуемым, документированным, обрабатывающим ошибки правильно, безопасным и масштабируемым.

Дизайн URL: основные принципы

Принципы именования ресурсов:

✅ Существительные во множественном числе: /api/v1/devices /api/v1/devices/42 /api/v1/devices/42/sensors /api/v1/devices/42/sensors/7/readings

❌ Глаголы в URL: /api/v1/getDevices ← нет! /api/v1/createDevice ← нет! /api/v1/device/42/delete ← нет! Действие выражается HTTP-методом: GET /devices → список устройств POST /devices → создать устройство GET /devices/42 → получить устройство 42 PUT /devices/42 → полностью заменить устройство 42 PATCH /devices/42 → частично обновить устройство 42 DELETE /devices/42 → удалить устройство 42 Вложенность — для зависимых ресурсов: GET /devices/42/sensors → датчики устройства 42 POST /devices/42/sensors → добавить датчик к устройству 42 GET /devices/42/sensors/7 → конкретный датчик DELETE /devices/42/sensors/7 → удалить датчик Максимум 3 уровня вложенности! Глубже — smell.

HTTP методы и их семантика

Идемпотентность: повторный вызов даёт тот же результат. PUT /devices/42 с одними данными можно вызвать 100 раз — результат одинаковый. DELETE /devices/42 — тоже (второй вызов: 404, но ресурс всё равно удалён).

Коды состояния HTTP: правильное использование

2xx — Успех:
  200 OK          — GET, PUT, PATCH успешно
  201 Created     — POST создал ресурс; Location: /api/v1/devices/43
  204 No Content  — DELETE, PATCH без возврата данных
  
3xx — Перенаправление:
  301 Moved Permanently — ресурс переехал навсегда
  304 Not Modified      — кешированный ответ актуален (ETag/If-None-Match)

4xx — Ошибка клиента:
  400 Bad Request       — невалидные данные запроса
  401 Unauthorized      — не аутентифицирован (нет или неверный токен)
  403 Forbidden         — аутентифицирован, но нет прав
  404 Not Found         — ресурс не существует
  405 Method Not Allowed— метод не поддерживается для этого URL
  409 Conflict          — конфликт (дубликат, версионирование)
  422 Unprocessable     — синтаксически корректный JSON, но семантически неверный
  429 Too Many Requests — rate limit превышен

5xx — Ошибка сервера:
  500 Internal Server Error — непредвиденная ошибка
  503 Service Unavailable   — временно недоступен (maintenance, overload)

Структура ответов: единообразие обязательно

# Стандартизированный формат ответа

# Успешный список с пагинацией:
{
    "data": [
        {"id": 1, "name": "Насос 1", "status": "active"},
        {"id": 2, "name": "Насос 2", "status": "fault"}
    ],
    "meta": {
        "total": 48,
        "page": 2,
        "page_size": 10,
        "pages": 5,
        "next": "/api/v1/devices?page=3&page_size=10",
        "prev": "/api/v1/devices?page=1&page_size=10"
    }
}

# Единичный объект:
{
    "data": {
        "id": 42,
        "name": "Насос холодной воды",
        "location": "Котельная",
        "created_at": "2024-01-15T10:30:00Z",
        "updated_at": "2024-03-10T08:45:22Z"
    }
}

# Ошибка (ВСЕГДА одна структура!):
{
    "error": {
        "code": "VALIDATION_ERROR",
        "message": "Ошибка валидации входных данных",
        "details": [
            {
                "field": "temperature_max",
                "message": "Значение должно быть больше temperature_min"
            },
            {
                "field": "device_id",
                "message": "Устройство с таким ID не существует"
            }
        ],
        "request_id": "req_abc123xyz",
        "timestamp": "2024-03-15T14:22:33Z"
    }
}

FastAPI: production-ready API за минуты

# main.py
from fastapi import FastAPI, HTTPException, Depends, Query, Path, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
import uuid

app = FastAPI(
    title="Industrial IoT API",
    description="API управления промышленными устройствами",
    version="1.0.0",
    docs_url="/api/docs",      # Swagger UI
    redoc_url="/api/redoc",    # ReDoc
    openapi_url="/api/openapi.json"
)

# Middleware
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://dashboard.factory.com"],
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
)

# ===== МОДЕЛИ =====

class DeviceStatus(str):
    ACTIVE   = "active"
    INACTIVE = "inactive"
    FAULT    = "fault"
    MAINTENANCE = "maintenance"

class DeviceBase(BaseModel):
    name:     str = Field(..., min_length=1, max_length=100, example="Насос 1")
    location: str = Field(..., example="Котельная")
    model:    str = Field(..., example="Grundfos CM5-5")
    tags:     List[str] = Field(default=[], example=["pump", "cooling"])

class DeviceCreate(DeviceBase):
    pass

class DeviceUpdate(BaseModel):
    name:     Optional[str] = Field(None, min_length=1, max_length=100)
    location: Optional[str] = None
    status:   Optional[str] = None
    tags:     Optional[List[str]] = None

class DeviceResponse(DeviceBase):
    id:         int
    status:     str = "active"
    created_at: datetime
    updated_at: datetime
    
    class Config:
        from_attributes = True

class PaginatedResponse(BaseModel):
    data:  List[DeviceResponse]
    meta:  dict

# ===== ЗАВИСИМОСТИ =====

# Имитация БД
fake_db = {}
device_counter = 0

def get_device_or_404(device_id: int = Path(..., ge=1)) -> dict:
    device = fake_db.get(device_id)
    if not device:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail={
                "error": {
                    "code": "DEVICE_NOT_FOUND",
                    "message": f"Устройство с ID {device_id} не найдено",
                    "request_id": str(uuid.uuid4())
                }
            }
        )
    return device

# ===== ENDPOINTS =====

@app.get(
    "/api/v1/devices",
    response_model=PaginatedResponse,
    summary="Список устройств",
    tags=["Devices"]
)
async def list_devices(
    page:      int   = Query(default=1, ge=1, description="Номер страницы"),
    page_size: int   = Query(default=20, ge=1, le=100, description="Размер страницы"),
    status:    Optional[str] = Query(default=None, description="Фильтр по статусу"),
    search:    Optional[str] = Query(default=None, description="Поиск по имени"),
    sort_by:   str   = Query(default="created_at", description="Поле сортировки"),
    sort_desc: bool  = Query(default=True, description="По убыванию"),
):
    """
    Возвращает список устройств с пагинацией и фильтрацией.
    
    - **page**: страница (начиная с 1)
    - **page_size**: количество на странице (макс. 100)
    - **status**: фильтр по статусу (active, inactive, fault, maintenance)
    - **search**: поиск по имени и местоположению
    """
    # Имитация запроса к БД
    all_devices = list(fake_db.values())
    
    if status:
        all_devices = [d for d in all_devices if d.get("status") == status]
    
    if search:
        q = search.lower()
        all_devices = [d for d in all_devices
                      if q in d.get("name", "").lower()
                      or q in d.get("location", "").lower()]
    
    total = len(all_devices)
    pages = (total + page_size - 1) // page_size
    
    start = (page - 1) * page_size
    items = all_devices[start:start + page_size]
    
    base_url = f"/api/v1/devices?page_size={page_size}"
    
    return {
        "data": items,
        "meta": {
            "total":     total,
            "page":      page,
            "page_size": page_size,
            "pages":     pages,
            "next": f"{base_url}&page={page+1}" if page < pages else None,
            "prev": f"{base_url}&page={page-1}" if page > 1 else None,
        }
    }


@app.get(
    "/api/v1/devices/{device_id}",
    response_model=DeviceResponse,
    summary="Получить устройство",
    tags=["Devices"],
    responses={404: {"description": "Устройство не найдено"}}
)
async def get_device(device: dict = Depends(get_device_or_404)):
    return device


@app.post(
    "/api/v1/devices",
    response_model=DeviceResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Создать устройство",
    tags=["Devices"]
)
async def create_device(device_data: DeviceCreate):
    global device_counter
    device_counter += 1
    
    now = datetime.utcnow()
    device = {
        "id":         device_counter,
        "status":     "active",
        "created_at": now,
        "updated_at": now,
        **device_data.dict()
    }
    
    fake_db[device_counter] = device
    
    return device


@app.patch(
    "/api/v1/devices/{device_id}",
    response_model=DeviceResponse,
    summary="Обновить устройство",
    tags=["Devices"]
)
async def update_device(
    update_data: DeviceUpdate,
    device: dict = Depends(get_device_or_404)
):
    # PATCH — обновляем только переданные поля
    updates = update_data.dict(exclude_unset=True)  # Только явно переданные поля!
    
    device.update(updates)
    device["updated_at"] = datetime.utcnow()
    
    return device


@app.delete(
    "/api/v1/devices/{device_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Удалить устройство",
    tags=["Devices"]
)
async def delete_device(device: dict = Depends(get_device_or_404)):
    fake_db.pop(device["id"])
    # 204 — нет тела ответа


# ===== ОБРАБОТЧИКИ ОШИБОК =====

@app.exception_handler(404)
async def not_found_handler(request, exc):
    return JSONResponse(
        status_code=404,
        content={"error": {"code": "NOT_FOUND", "message": "Ресурс не найден"}}
    )

@app.exception_handler(500)
async def server_error_handler(request, exc):
    # Логируем, но не раскрываем детали клиенту!
    import logging
    logging.exception(f"Unhandled error: {exc}")
    
    return JSONResponse(
        status_code=500,
        content={
            "error": {
                "code": "INTERNAL_ERROR",
                "message": "Внутренняя ошибка сервера",
                "request_id": str(uuid.uuid4())
            }
        }
    )

JWT авторизация

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from datetime import datetime, timedelta

SECRET_KEY  = "super-secret-key-change-in-production"
ALGORITHM   = "HS256"
TOKEN_EXPIRE = 60 * 24  # минуты

security = HTTPBearer()

def create_access_token(data: dict, expire_minutes: int = TOKEN_EXPIRE) -> str:
    payload = {
        **data,
        "exp": datetime.utcnow() + timedelta(minutes=expire_minutes),
        "iat": datetime.utcnow(),
        "type": "access"
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)) -> dict:
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY,
                             algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail={
            "error": {"code": "TOKEN_EXPIRED", "message": "Токен истёк"}
        })
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail={
            "error": {"code": "INVALID_TOKEN", "message": "Неверный токен"}
        })

# Использование:
@app.get("/api/v1/protected")
async def protected_route(current_user: dict = Depends(verify_token)):
    return {"user_id": current_user["sub"], "message": "OK"}

@app.post("/api/v1/auth/login")
async def login(username: str, password: str):
    # Проверка пользователя (в реальности — из БД)
    if username != "admin" or password != "secret":
        raise HTTPException(status_code=401,
                           detail={"error": {"code": "INVALID_CREDENTIALS"}})
    
    token = create_access_token({"sub": username, "role": "admin"})
    return {"access_token": token, "token_type": "bearer", "expires_in": TOKEN_EXPIRE * 60}

Rate Limiting

from fastapi import Request
from collections import defaultdict
import time

class InMemoryRateLimiter:
    """Простой rate limiter (для production используйте Redis)"""
    
    def __init__(self, requests_per_minute: int = 60):
        self.rpm   = requests_per_minute
        self.store = defaultdict(list)
    
    def is_allowed(self, key: str) -> tuple[bool, dict]:
        now = time.time()
        window = 60.0
        
        # Очищаем устаревшие записи
        self.store[key] = [t for t in self.store[key] if now - t < window]
        
        count = len(self.store[key])
        
        headers = {
            "X-RateLimit-Limit":     str(self.rpm),
            "X-RateLimit-Remaining": str(max(0, self.rpm - count - 1)),
            "X-RateLimit-Reset":     str(int(now + window)),
        }
        
        if count >= self.rpm:
            return False, headers
        
        self.store[key].append(now)
        return True, headers

rate_limiter = InMemoryRateLimiter(requests_per_minute=100)

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # Ключ: IP + User-Agent (или user_id после авторизации)
    client_ip = request.client.host
    key = f"{client_ip}:{request.headers.get('user-agent', '')[:50]}"
    
    allowed, headers = rate_limiter.is_allowed(key)
    
    if not allowed:
        return JSONResponse(
            status_code=429,
            content={"error": {"code": "RATE_LIMIT_EXCEEDED",
                               "message": "Слишком много запросов. Повторите через 60 секунд."}},
            headers={**headers, "Retry-After": "60"}
        )
    
    response = await call_next(request)
    response.headers.update(headers)
    return response

Версионирование API

# Способ 1: В URL (самый простой и видимый)
# /api/v1/devices
# /api/v2/devices

# Способ 2: В заголовке
# Accept: application/vnd.myapi.v2+json

# Способ 3: В параметре запроса
# /api/devices?version=2

# Рекомендуется: URL-версионирование для публичных API

# Правила обратной совместимости:
# 

✅ МОЖНО добавлять новые поля в ответ (клиент игнорирует незнакомые) #

✅ МОЖНО добавлять новые необязательные параметры #

✅ МОЖНО добавлять новые endpoints #

❌ НЕЛЬЗЯ удалять поля из ответа #

❌ НЕЛЬЗЯ менять тип поля #

❌ НЕЛЬЗЯ делать необязательный параметр обязательным #

❌ НЕЛЬЗЯ менять семантику существующих полей #

Любое из запрещённого → новая мажорная версия! from fastapi import APIRouter # v1 router v1_router = APIRouter(prefix="/api/v1", tags=["v1"]) @v1_router.get("/devices") async def list_devices_v1(): return {"version": "v1", "data": []} # v2 router (новая логика, breaking changes) v2_router = APIRouter(prefix="/api/v2", tags=["v2"]) @v2_router.get("/devices") async def list_devices_v2(): return {"version": "v2", "data": [], "meta": {}} # Новый формат app.include_router(v1_router) app.include_router(v2_router)

Документация: OpenAPI и Swagger

FastAPI автоматически генерирует OpenAPI spec. Добавьте подробные комментарии:

@app.get(
    "/api/v1/devices/{device_id}/telemetry",
    tags=["Telemetry"],
    summary="Телеметрия устройства",
    description="""
    Возвращает исторические данные телеметрии устройства.
    
    ## Параметры времени
    
    Используйте ISO 8601 формат: `2024-03-15T10:30:00Z`
    
    Или относительные значения: `-1h`, `-24h`, `-7d`, `-30d`
    
    ## Агрегация
    
    - `raw`: сырые данные (ограничено 10 000 точек)
    - `1min`: агрегация по минутам
    - `5min`, `1h`, `1d`: другие интервалы
    """,
    response_description="Список точек телеметрии с временными метками",
    responses={
        200: {"description": "Успешно"},
        404: {"description": "Устройство не найдено"},
        400: {"description": "Неверные параметры времени"},
    }
)
async def get_telemetry(
    device_id: int = Path(..., description="ID устройства", example=42),
    from_time: str = Query(..., description="Начало периода (ISO 8601 или -Nh/-Nd)", example="-24h"),
    to_time:   str = Query(default="now", description="Конец периода"),
    resample:  str = Query(default="5min", description="Интервал агрегации",
                           regex="^(raw|1min|5min|1h|1d)$"),
):
    ...

Чеклист для production API

Безопасность:
□ HTTPS только (HTTP редирект на HTTPS)
□ JWT с разумным TTL (1-24 часа)
□ Rate limiting по IP и по user
□ Валидация всех входных данных
□ SQL-инъекции: параметризованные запросы (ORM)
□ Секреты не в коде (env vars, Vault)
□ CORS настроен правильно (не allow_origins=["*"]!)

Надёжность:
□ Timeouts на все внешние запросы
□ Circuit Breaker для сервисов-зависимостей
□ Graceful shutdown (SIGTERM обработан)
□ Health endpoint /health или /api/health

Observability:
□ Структурированные логи (JSON) с request_id
□ Метрики: latency, error rate, throughput
□ Distributed tracing (OpenTelemetry)

Документация:
□ OpenAPI/Swagger автоматически
□ Примеры запросов в описаниях
□ CHANGELOG с версиями

Заключение

Хороший REST API — это API, которым приятно пользоваться. Предсказуемые URL, понятные коды ошибок, единообразная структура ответов, версионирование без сюрпризов, документация которая не врёт.

FastAPI делает большую часть работы автоматически: валидацию, сериализацию, документацию. Но архитектурные решения — за вами. Потратьте время на проектирование URL до написания кода. Нарисуйте ресурсы и операции. Согласуйте с командой стандарты ошибок. Это окупится при первом обращении клиентской команды к вашему API.