Dependency Injection в FastAPI

Depends — система внедрения зависимостей FastAPI для переиспользования кода.

Базовый пример

from fastapi import FastAPI, Depends

app = FastAPI()

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

@app.get("/tasks/")
def list_tasks(db: Session = Depends(get_db)):
    return db.query(Task).all()

Параметры запроса как зависимость

class Pagination:
    def __init__(self, page: int = 1, size: int = 20):
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

@app.get("/tasks/")
def list_tasks(pagination: Pagination = Depends()):
    return {
        "page": pagination.page,
        "offset": pagination.offset,
    }

Аутентификация через Depends

from fastapi import HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    user = verify_token(token, db)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

@app.get("/tasks/")
def list_tasks(current_user: User = Depends(get_current_user)):
    return db.query(Task).filter(Task.owner_id == current_user.id).all()

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

def get_db():
    ...  # зависимость 1

def get_current_user(db = Depends(get_db)):
    ...  # зависимость 2, использует 1

def get_admin_user(user = Depends(get_current_user)):
    if not user.is_admin:
        raise HTTPException(403)
    return user  # зависимость 3, использует 2

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int, admin: User = Depends(get_admin_user)):
    ...

yield зависимости (с cleanup)

def get_db():
    db = SessionLocal()
    try:
        yield db        # код до yield выполняется до хендлера
    finally:
        db.close()      # код после yield — после хендлера (cleanup)

Depends на уровне роутера

router = APIRouter(
    prefix="/tasks",
    dependencies=[Depends(get_current_user)],  # для всех endpoints роутера
)