Приведи пример использования Depends

Пример использования Depends в FastAPI

Depends — это инструмент FastAPI для инъекции зависимостей. Он позволяет переиспользовать код, автоматически вызывая функции и внедряя их результаты в обработчики маршрутов.

Базовый пример: аутентификация

from fastapi import FastAPI, Depends, HTTPException, status
from typing import Optional

app = FastAPI()

# Функция-зависимость для получения текущего пользователя
def get_current_user(token: Optional[str] = None) -> dict:
    """Извлекает текущего пользователя из токена"""
    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Токен не предоставлен"
        )
    
    # Валидация токена
    if token != "valid-token-123":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Неверный токен"
        )
    
    return {"user_id": 1, "name": "John Doe", "email": "john@example.com"}

# Используем Depends в маршруте
@app.get("/me")
async def get_profile(current_user: dict = Depends(get_current_user)):
    """Получить профиль текущего пользователя"""
    return {
        "message": f"Привет, {current_user['name']}",
        "user": current_user
    }

@app.get("/secure")
async def secure_endpoint(current_user: dict = Depends(get_current_user)):
    """Защищённый endpoint, требует аутентификации"""
    return {"secure_data": "Only authenticated users see this", "user": current_user}

Практический пример: проверка прав доступа

from enum import Enum

class Role(str, Enum):
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

# Симуляция базы данных
USERS_DB = {
    "valid-token-123": {"user_id": 1, "name": "Admin", "role": Role.ADMIN},
    "user-token-456": {"user_id": 2, "name": "Regular User", "role": Role.USER},
}

def get_current_user(token: str) -> dict:
    """Получить текущего пользователя"""
    user = USERS_DB.get(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )
    return user

def require_admin(current_user: dict = Depends(get_current_user)) -> dict:
    """Проверить, что пользователь администратор"""
    if current_user["role"] != Role.ADMIN:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    return current_user

# Использование в маршрутах
@app.get("/admin/users")
async def list_all_users(admin: dict = Depends(require_admin)):
    """Только администратор может получить список всех пользователей"""
    return {
        "message": f"Запрос от администратора {admin['name']}",
        "users": list(USERS_DB.values())
    }

@app.delete("/users/{user_id}")
async def delete_user(user_id: int, admin: dict = Depends(require_admin)):
    """Удалить пользователя (только администратор)"""
    return {
        "message": f"Пользователь {user_id} удалён администратором {admin['name']}"
    }

@app.get("/user-profile")
async def get_user_profile(current_user: dict = Depends(get_current_user)):
    """Получить свой профиль (любой аутентифицированный пользователь)"""
    return {"profile": current_user}

Цепочка зависимостей

# Зависимость от зависимости

def verify_token(token: str) -> str:
    """Проверить и вернуть токен"""
    if len(token) < 5:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Token too short"
        )
    return token

def get_user_from_token(token: str = Depends(verify_token)) -> dict:
    """Получить пользователя из проверенного токена"""
    user = USERS_DB.get(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return user

def log_user_action(user: dict = Depends(get_user_from_token)):
    """Логировать действие пользователя"""
    print(f"User {user['user_id']} performed an action")
    return user

@app.post("/action")
async def perform_action(user: dict = Depends(log_user_action)):
    """Выполнить действие с логированием"""
    return {"status": "success", "user": user['name']}

Кэширование зависимостей

from sqlalchemy.orm import Session

def get_db() -> Session:
    """Получить сессию БД (кэшируется для одного запроса)"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/profile/{user_id}")
async def get_user_profile(
    user_id: int,
    db: Session = Depends(get_db)  # Одна сессия на весь запрос
):
    """Получить профиль пользователя из БД"""
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Множественные зависимости

def get_query_params(
    skip: int = 0,
    limit: int = 10
) -> dict:
    """Получить параметры пагинации из query"""
    return {"skip": skip, "limit": limit}

def get_sort_params(
    sort_by: str = "id",
    order: str = "asc"
) -> dict:
    """Получить параметры сортировки"""
    return {"sort_by": sort_by, "order": order}

@app.get("/items")
async def list_items(
    pagination: dict = Depends(get_query_params),
    sorting: dict = Depends(get_sort_params),
    current_user: dict = Depends(get_current_user)
):
    """Получить список элементов с пагинацией и сортировкой"""
    return {
        "pagination": pagination,
        "sorting": sorting,
        "user": current_user["name"],
        "items": []  # Результаты из БД
    }

Класс как зависимость

class PaginationParams:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit
        
        # Валидация
        if self.skip < 0:
            raise HTTPException(status_code=400, detail="skip must be >= 0")
        if self.limit > 100:
            raise HTTPException(status_code=400, detail="limit must be <= 100")

@app.get("/users")
async def list_users(
    pagination: PaginationParams = Depends(),
    current_user: dict = Depends(get_current_user)
):
    """Получить список пользователей с валидированной пагинацией"""
    return {
        "pagination": {
            "skip": pagination.skip,
            "limit": pagination.limit
        },
        "requester": current_user["name"]
    }

Реальный пример: REST API

from sqlalchemy.orm import Session
from typing import List

class Repository:
    """Репозиторий для работы с БД"""
    def __init__(self, db: Session):
        self.db = db
    
    def get_user(self, user_id: int) -> Optional[dict]:
        # SELECT * FROM users WHERE id = user_id
        pass
    
    def update_user(self, user_id: int, data: dict) -> dict:
        # UPDATE users SET ... WHERE id = user_id
        pass

def get_repository(db: Session = Depends(get_db)) -> Repository:
    """Получить репозиторий"""
    return Repository(db)

@app.put("/users/{user_id}")
async def update_user(
    user_id: int,
    data: dict,
    repo: Repository = Depends(get_repository),
    current_user: dict = Depends(get_current_user)
):
    """Обновить данные пользователя"""
    # Проверка прав
    if current_user["user_id"] != user_id and current_user["role"] != "admin":
        raise HTTPException(status_code=403, detail="Forbidden")
    
    updated = repo.update_user(user_id, data)
    return {"status": "updated", "user": updated}

Ключевые преимущества Depends

# 1. Переиспользование кода
def get_current_user(token: str) -> dict:
    # Один раз написали, используем везде
    pass

# 2. Тестируемость
def test_endpoint():
    fake_user = {"user_id": 1}
    result = endpoint(user=fake_user)  # Легко мокировать
    assert result["status"] == "success"

# 3. Разделение ответственности
# Валидация в get_current_user
# Бизнес-логика в endpoint

# 4. Нет "паразитов" в параметрах функции
# Вместо: async def endpoint(token, skip, limit, ...)
# Используем: async def endpoint(current_user=Depends(...))

Вывод

Depends в FastAPI — это паттерн инъекции зависимостей. Он позволяет:

• Переиспользовать код
• Разделить ответственность
• Упростить тестирование
• Сделать код чище и понятнее

Это особенно полезно для аутентификации, авторизации, работы с БД и валидации параметров.