Как в FastAPI делается инверсия зависимостей?

Инверсия зависимостей в FastAPI

FastAPI использует Dependency Injection (DI) систему для реализации инверсии управления (IoC). Это позволяет автоматически внедрять зависимости в обработчики маршрутов.

Базовый механизм

FastAPI анализирует сигнатуру функции-обработчика и автоматически создаёт и внедряет необходимые зависимости:

from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

# 1. Определяем функцию-зависимость
def get_database_url() -> str:
    return "postgresql://user:pass@localhost/db"

# 2. Внедряем зависимость через Depends()
@app.get("/items")
def read_items(db_url: Annotated[str, Depends(get_database_url)]):
    return {"db": db_url}

# GET /items → FastAPI вызывает get_database_url() → передаёт результат в обработчик

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

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

from fastapi import FastAPI, Depends, HTTPException
from typing import Annotated

app = FastAPI()

# Уровень 1: базовая зависимость
def get_db_connection():
    connection = {"connected": True}
    print("Открыли соединение с БД")
    yield connection
    print("Закрыли соединение с БД")

# Уровень 2: зависит от уровня 1
def get_current_user(
    db: Annotated[dict, Depends(get_db_connection)]
) -> dict:
    user = {"id": 1, "name": "Alice"}
    print(f"Загрузили пользователя из БД: {user}")
    return user

# Уровень 3: зависит от уровня 2
@app.get("/protected")
def protected_route(
    user: Annotated[dict, Depends(get_current_user)]
):
    return {"message": f"Hello, {user[name]}"}

При запросе к /protected FastAPI выполнит цепочку: открыть БД → загрузить пользователя → обработать маршрут → закрыть БД.

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

Без кэширования одна зависимость может вызваться несколько раз. FastAPI кэширует её в рамках одного запроса:

from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

call_count = 0

def expensive_operation() -> dict:
    global call_count
    call_count += 1
    print(f"Дорогая операция #{call_count}")
    return {"result": "data"}

@app.get("/test")
def test_route(
    result1: Annotated[dict, Depends(expensive_operation)],
    result2: Annotated[dict, Depends(expensive_operation)]
):
    # expensive_operation вызовется 1 раз, результат переиспользуется
    return {"result1": result1, "result2": result2}

Eсли нужно отключить кэширование:

@app.get("/test")
def test_route(
    result1: Annotated[dict, Depends(expensive_operation, use_cache=False)],
    result2: Annotated[dict, Depends(expensive_operation, use_cache=False)]
):
    # expensive_operation вызовется 2 раза
    return {"result1": result1, "result2": result2}

Контекстные менеджеры (yield)

Для cleanup логики используются генераторы с yield:

from fastapi import FastAPI, Depends
from typing import Annotated, Generator

app = FastAPI()

def get_db() -> Generator[dict, None, None]:
    print("Подключение к БД")
    db = {"connection": "active"}
    try:
        yield db
    finally:
        print("Отключение от БД")
        db["connection"] = "closed"

@app.get("/data")
def get_data(db: Annotated[dict, Depends(get_db)]):
    return db

# Порядок выполнения:
# 1. print("Подключение к БД")
# 2. yield db
# 3. GET /data → db используется в обработчике
# 4. Возврат ответа
# 5. finally: print("Отключение от БД")

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

Обычно используют классы для state и dependency injection:

from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

class DatabaseService:
    def __init__(self):
        self.url = "postgresql://localhost/db"
    
    def query(self, sql: str):
        return f"Результат: {sql}"

class AuthService:
    def __init__(self, db: DatabaseService):
        self.db = db
    
    def authenticate(self, username: str) -> dict:
        result = self.db.query(f"SELECT * FROM users WHERE name={username}")
        return {"user": username, "authenticated": True}

# FastAPI автоматически инстанцирует классы в правильном порядке
@app.get("/login/{username}")
def login(
    auth: Annotated[AuthService, Depends()]
):
    user = auth.authenticate("alice")
    return user

FastAPI проанализирует сигнатуру AuthService.__init__() и автоматически создаст DatabaseService перед созданием AuthService.

Обработка ошибок в зависимостях

from fastapi import FastAPI, Depends, HTTPException
from typing import Annotated

app = FastAPI()

def verify_token(token: str) -> dict:
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="Неверный токен")
    return {"token": token, "valid": True}

@app.get("/protected")
def protected(
    auth: Annotated[dict, Depends(verify_token)]
):
    return {"message": "Доступ разрешён"}

# Если зависимость выбросит исключение, обработчик маршрута не вызовется

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

from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

def log_request(request_id: str = "") -> str:
    return f"req-{request_id}"

# Применяет зависимость ко ВСЕМ маршрутам
app = FastAPI(dependencies=[Depends(log_request)])

@app.get("/")
def read_root():
    return {"message": "Hello"}

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

1. Тестируемость — легко подменять реальные зависимости на моки
2. Переиспользование кода — одна зависимость для множества маршрутов
3. Чистота — обработчики маршрутов сосредоточены на бизнес-логике
4. Автоматическая документация — OpenAPI включает информацию о зависимостях

# Для тестов подменяем реальные зависимости
from fastapi.testclient import TestClient

app = FastAPI()

def real_db():
    return {"type": "real"}

fake_db = {"type": "fake"}

app.dependency_overrides[real_db] = lambda: fake_db

client = TestClient(app)
response = client.get("/test")

Inversion of Control в FastAPI реализована через систему Depends(), которая автоматически разрешает и внедряет зависимости, позволяя создавать гибкий, тестируемый и масштабируемый код.