Приведи пример dependency в FastAPI

Dependency Injection в FastAPI

Dependency Injection (DI) — это паттерн, позволяющий внедрять зависимости в функции-обработчики. FastAPI имеет встроенную систему DI.

Базовый пример: зависимость параметров запроса

from fastapi import FastAPI, Depends

app = FastAPI()

# 1. Определяем функцию-зависимость
def common_parameters(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 2. Используем её через Depends()
@app.get("/items")
async def list_items(params = Depends(common_parameters)):
    return {
        "items": [f"Item {i}" for i in range(params["skip"], params["skip"] + params["limit"])],
        "skip": params["skip"],
        "limit": params["limit"]
    }

Пример 2: Аутентификация

from fastapi import HTTPException
from fastapi.security import HTTPBearer, HTTPAuthenticationCredentials
import jwt

security = HTTPBearer()
SECRET_KEY = "your-secret-key"

class User:
    def __init__(self, user_id: int, username: str, role: str):
        self.id = user_id
        self.username = username
        self.role = role

async def get_current_user(credentials: HTTPAuthenticationCredentials = Depends(security)) -> User:
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")
        username = payload.get("username")
        role = payload.get("role")
        
        if not user_id:
            raise HTTPException(status_code=401, detail="Invalid token")
        
        return User(int(user_id), username, role)
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid credentials")

@app.get("/api/v1/users/me")
async def get_profile(current_user: User = Depends(get_current_user)):
    return {
        "id": current_user.id,
        "username": current_user.username,
        "role": current_user.role
    }

Пример 3: Подключение к БД

from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/api/v1/users/{user_id}")
async def get_user(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 {"id": user.id, "username": user.username}

Пример 4: Комбинирование зависимостей

async def get_current_user(token: str = Header(...)) -> User:
    return User(id=1, username="john", role="admin")

async def require_admin(current_user: User = Depends(get_current_user)) -> User:
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return current_user

@app.post("/api/v1/users")
async def create_user(
    user_data: dict,
    admin: User = Depends(require_admin),
    db: Session = Depends(get_db)
):
    new_user = User(**user_data)
    db.add(new_user)
    db.commit()
    return {"id": new_user.id, "created": True}

Преимущества DI

✅ Повторное использование — одна зависимость в разных обработчиках ✅ Тестирование — легко подменять зависимости на mock'и ✅ Чистота кода — обработчики сосредоточены на бизнес-логике ✅ Разделение ответственности — каждая зависимость отвечает за одно

Dependency Injection в FastAPI — мощный механизм для организации кода и повышения тестируемости приложения.