Dependency Injection — FastAPI Pro: От новичка до

Dependency Injection (DI) в FastAPI

Dependency Injection — мощный паттерн для переиспользования кода, разделения ответственности и чистой архитектуры. В этой теме вы научитесь использовать Depends() для внедрения зависимостей.

#Что такое Dependency Injection?

Dependency Injection (внедрение зависимостей) — паттерн, при котором зависимости (сервисы, репозитории, конфигурация) передаются в объект извне, а не создаются внутри.

#Без DI (плохо)

@app.get('/users/{user_id}')
def get_user(user_id: int):
    db = SessionLocal()  # Создаём зависимость внутри
    try:
        user = db.query(User).filter(User.id == user_id).first()
        return user
    finally:
        db.close()

Проблемы:

Код создания БД дублируется в каждом эндпоинте
Сложно тестировать (нужно создавать реальную БД)
Невозможно легко подменить реализацию

#С DI (хорошо)

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

@app.get('/users/{user_id}')
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    return user

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

Логика подключения БД в одном месте
Легко тестировать (можно подменить get_db)
Чистый код эндпоинтов

#Depends() — основа DI в FastAPI

Depends() говорит FastAPI: «Получи эту зависимость и передай в функцию».

from fastapi import FastAPI, Depends

app = FastAPI()

def get_common_params(q: str | None = None, skip: int = 0, limit: int = 100):
    return {'q': q, 'skip': skip, 'limit': limit}

@app.get('/items')
def list_items(params: dict = Depends(get_common_params)):
    return params

FastAPI:

Вызывает get_common_params() с query-параметрами
Получает результат
Передаёт в list_items как параметр params

#Зависимости с yield

Для ресурсов, требующих очистки (БД, соединения), используется yield:

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db  # Передаём db в эндпоинт
    finally:
        db.close()  # Закрываем после выполнения

@app.get('/users', response_model=list[UserResponse])
def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()

#Жизненный цикл

1. Запрос приходит
        ↓
2. FastAPI вызывает get_db()
        ↓
3. get_db() создаёт db, делает yield db
        ↓
4. FastAPI передаёт db в list_users()
        ↓
5. list_users() выполняется, возвращает ответ
        ↓
6. FastAPI возобновляет get_db() после yield
        ↓
7. get_db() закрывает соединение (finally)
        ↓
8. Ответ отправляется клиенту

#Зависимости в зависимостях

Зависимости могут использовать другие зависимости:

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

def get_current_user(db: Session = Depends(get_db)):
    # Получаем токен из заголовков
    token = get_token_from_headers()
    user = db.query(User).filter(User.token == token).first()
    
    if not user:
        raise HTTPException(status_code=401, detail="Not authenticated")
    
    return user

def get_current_active_user(
    current_user: User = Depends(get_current_user)
):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    
    return current_user

@app.get('/users/me')
def read_current_user(
    user: User = Depends(get_current_active_user)
):
    return user

#Цепочка вызовов

read_current_user
    ↓ Depends(get_current_active_user)
get_current_active_user
    ↓ Depends(get_current_user)
get_current_user
    ↓ Depends(get_db)
get_db (yield db)

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

#Классы как зависимости

Можно использовать классы вместо функций:

class Pagination:
    def __init__(self, page: int = 1, page_size: int = 10):
        self.page = page
        self.page_size = page_size
        self.skip = (page - 1) * page_size

@app.get('/items')
def list_items(pagination: Pagination = Depends()):
    items = db.query(Item).offset(pagination.skip).limit(pagination.page_size).all()
    return {'items': items, 'page': pagination.page}

FastAPI создаст экземпляр Pagination, передав query-параметры в __init__.

#Зависимости для аутентификации

Классический случай DI — проверка аутентификации:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt
from jwt.exceptions import InvalidTokenError

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

SECRET_KEY = "secret"
ALGORITHM = "HS256"

def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        
        if username is None:
            raise credentials_exception
        
    except InvalidTokenError:
        raise credentials_exception
    
    user = get_user_from_db(username)
    
    if user is None:
        raise credentials_exception
    
    return user

@app.get('/users/me')
def read_current_user(current_user: User = Depends(get_current_user)):
    return current_user

#Использование в нескольких эндпоинтах

@app.get('/users/me')
def read_current_user(current_user: User = Depends(get_current_user)):
    return current_user

@app.put('/users/me')
def update_current_user(
    user_update: UserUpdate,
    current_user: User = Depends(get_current_user)
):
    return update_user_in_db(current_user.id, user_update)

@app.delete('/users/me', status_code=204)
def delete_current_user(
    current_user: User = Depends(get_current_user)
):
    delete_user_from_db(current_user.id)

#Зависимости с параметрами

Можно передавать параметры в зависимости:

def get_pagination(page: int = 1, page_size: int = 10):
    skip = (page - 1) * page_size
    return {'skip': skip, 'limit': page_size}

@app.get('/items')
def list_items(pagination: dict = Depends(get_pagination)):
    items = db.query(Item).offset(pagination['skip']).limit(pagination['limit']).all()
    return items

#Зависимость с аргументами

def get_filter(model_class: type, field: str, value: str):
    def filter_func(db: Session = Depends(get_db)):
        return db.query(model_class).filter(getattr(model_class, field) == value).first()
    return filter_func

@app.get('/users/{username}')
def get_user(user: User = Depends(lambda: get_filter(User, 'username', ...))):
    return user

#Глобальные зависимости

Можно применить зависимость ко всему приложению:

from fastapi import FastAPI, Depends

app = FastAPI(dependencies=[Depends(verify_token)])

async def verify_token():
    # Проверка токена для всех эндпоинтов
    token = get_token()
    if not is_valid(token):
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get('/items')
def list_items():
    return db.query(Item).all()  # Токен уже проверен

@app.get('/users')
def list_users():
    return db.query(User).all()  # Токен уже проверен

#Зависимости для роутеров

Можно применить зависимость к конкретному роутеру:

from fastapi import APIRouter, Depends

router = APIRouter(dependencies=[Depends(get_current_user)])

@router.get('/items')
def list_items():
    ...

@router.get('/users')
def list_users():
    ...

app.include_router(router, prefix='/api')

Все эндпоинты в роутере требуют аутентификации.

#Пример: Полноценная система DI

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import Annotated

app = FastAPI()

# === База данных ===
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

DbDependency = Annotated[Session, Depends(get_db)]

# === Аутентификация ===
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: DbDependency
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        
        if username is None:
            raise credentials_exception
        
    except InvalidTokenError:
        raise credentials_exception
    
    user = db.query(User).filter(User.username == username).first()
    
    if user is None:
        raise credentials_exception
    
    return user

UserDependency = Annotated[User, Depends(get_current_user)]

def get_current_active_user(current_user: UserDependency):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

ActiveUserDependency = Annotated[User, Depends(get_current_active_user)]

def get_current_admin_user(current_user: UserDependency):
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Not enough permissions"
        )
    return current_user

AdminUserDependency = Annotated[User, Depends(get_current_admin_user)]

# === Пагинация ===
def get_pagination(page: int = 1, page_size: int = 20):
    skip = (page - 1) * page_size
    return {'skip': skip, 'limit': page_size}

PaginationDependency = Annotated[dict, Depends(get_pagination)]

# === Endpoints ===
@app.get('/users/me')
def read_current_user(current_user: ActiveUserDependency):
    return current_user

@app.get('/users')
def list_users(
    current_user: AdminUserDependency,
    pagination: PaginationDependency
):
    users = db.query(User).offset(pagination['skip']).limit(pagination['limit']).all()
    return users

@app.post('/posts')
def create_post(
    post: PostCreate,
    current_user: ActiveUserDependency,
    db: DbDependency
):
    new_post = Post(**post.model_dump(), author_id=current_user.id)
    db.add(new_post)
    db.commit()
    db.refresh(new_post)
    return new_post

@app.delete('/posts/{post_id}')
def delete_post(
    post_id: int,
    current_user: AdminUserDependency,
    db: DbDependency
):
    post = db.query(Post).filter(Post.id == post_id).first()
    
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    
    db.delete(post)
    db.commit()
    return {'deleted': True}

#Тестирование с DI

DI позволяет легко подменять зависимости в тестах:

from fastapi.testclient import TestClient

# Тестовая БД
def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_read_users():
    response = client.get("/users")
    assert response.status_code == 200

#Частые ошибки

#Ошибка 1: Зависимость не вызывается

@app.get('/items')
def list_items(db: Session = Depends(get_db)):  # Забыли ()
    ...

Проблема: Depends(get_db) вместо Depends(get_db()) — неправильно, но работает, потому что FastAPI понимает оба варианта.

Решение: Используйте Depends(get_db) — это предпочтительный синтаксис.

#Ошибка 2: Yield без finally

def get_db():
    db = SessionLocal()
    yield db
    # Забыли db.close()!

Проблема: Соединение не закроется при ошибке.

Решение:

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

#Ошибка 3: Круговая зависимость

def dep_a(dep_b = Depends(get_dep_b)): ...
def dep_b(dep_a = Depends(get_dep_a)): ...

Проблема: Бесконечный цикл.

Решение: Перестройте зависимости, чтобы не было циклов.

#Исполняемая лаборатория

Лаборатория «Depends, yield-зависимости, фильтрация по роли» доступна в публичном GitLab. Она проверяет цепочки зависимостей, yield с очисткой ресурсов, Annotated[..., Depends(...)], dependency_overrides в тестах и проверку роли через DI.

Склонируйте репозиторий и запустите тесты из его корня:

git clone --branch v1.0 --depth 1 https://gitlab.potapov.me/courses/fastapi_pro.git
cd fastapi_pro
uv sync --group test
uv run --group test pytest dependency_injection/tests

1. Запрос приходит ↓ 2. FastAPI вызывает get_db() ↓ 3. get_db() создаёт db, делает yield db ↓ 4. FastAPI передаёт db в list_users() ↓ 5. list_users() выполняется, возвращает ответ ↓ 6. FastAPI возобновляет get_db() после yield ↓ 7. get_db() закрывает соединение (finally) ↓ 8. Ответ отправляется клиенту