Что такое передача зависимостей FastAPI?

Передача зависимостей (Dependency Injection) в FastAPI

Передача зависимостей (Dependency Injection — DI) в FastAPI — это встроенный механизм для внедрения зависимостей в обработчики маршрутов. Это позволяет полностью избежать глобального состояния, делает тесты проще и код чище, а также обеспечивает переиспользуемость.

Вместо того чтобы инициализировать всё необходимое внутри функции-обработчика, вы объявляете параметры функции как зависимости, и FastAPI автоматически передаёт их.

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

from fastapi import FastAPI, Depends

app = FastAPI()

def get_query_param(q: str = None):
    return q

@app.get("/search")
async def search(q: str = Depends(get_query_param)):
    return {"query": q}

Здесь get_query_param — функция-зависимость. FastAPI вызовет её и передаст результат в параметр q обработчика.

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

from fastapi import FastAPI, Depends

app = FastAPI()

def verify_token(token: str):
    if token != "secret-token":
        raise Exception("Invalid token")
    return token

@app.get("/secure")
async def secure_endpoint(token: str, verified_token: str = Depends(verify_token)):
    return {"message": "Access granted", "token": verified_token}

Зависимость verify_token получает параметр token из query-параметров или request body.

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

from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
    db = connect_to_database()
    yield db
    db.close()

def get_current_user(db = Depends(get_db), token: str = None):
    return db.query_user(token)

@app.get("/profile")
async def get_profile(current_user = Depends(get_current_user)):
    return {"user": current_user}

Зависимость get_current_user зависит от get_db, которая зависит от параметров запроса. FastAPI разрешает всю цепочку автоматически.

Генераторы (контекст-менеджеры) для управления ресурсами

from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager

app = FastAPI()

async def get_db():
    db = await connect_async_db()
    try:
        yield db
    finally:
        await db.close()

@app.get("/users")
async def get_users(db = Depends(get_db)):
    users = await db.fetch_users()
    return {"users": users}

У yield — эта часть выполняется ДО обработчика, после yield — ПОСЛЕ. Идеально для открытия/закрытия соединений, транзакций и т.д.

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

from fastapi import FastAPI, Depends

app = FastAPI()

class DatabaseService:
    def __init__(self, connection_string: str = "postgresql://localhost"):
        self.connection_string = connection_string
        self.db = self.connect()
    
    def connect(self):
        return "connected"
    
    def get_user(self, user_id: int):
        return {"id": user_id, "name": "John"}

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: DatabaseService = Depends()):
    return db.get_user(user_id)

Можно передать класс как зависимость напрямую. FastAPI создаст экземпляр класса автоматически.

Зависимости из таблицы параметров

from fastapi import FastAPI, Depends, Query
from typing import Optional

app = FastAPI()

class FilterParams:
    def __init__(
        self,
        skip: int = Query(0, ge=0),
        limit: int = Query(10, ge=1, le=100),
        sort: Optional[str] = None
    ):
        self.skip = skip
        self.limit = limit
        self.sort = sort

@app.get("/items")
async def list_items(params: FilterParams = Depends()):
    return {"skip": params.skip, "limit": params.limit, "sort": params.sort}

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

from fastapi import FastAPI, Depends

def verify_api_key(api_key: str):
    if api_key != "valid-key":
        raise Exception("Invalid API key")

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

@app.get("/protected")
async def protected_route():
    # api_key проверяется для всех маршрутов
    return {"message": "Protected resource"}

Зависимости с бизнес-логикой

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

class UserService:
    def __init__(self, db = Depends(get_db)):
        self.db = db
    
    def get_user_or_404(self, user_id: int):
        user = self.db.get(user_id)
        if not user:
            raise HTTPException(status_code=404, detail="User not found")
        return user

@app.get("/users/{user_id}")
async def get_user(user_id: int, service: UserService = Depends()):
    return service.get_user_or_404(user_id)

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

По умолчанию FastAPI кэширует результаты зависимостей в рамках одного запроса. Если одна зависимость используется несколько раз — она вычисляется один раз:

from fastapi import FastAPI, Depends

def expensive_operation():
    print("Computing...")
    return {"result": 42}

@app.get("/data")
async def get_data(data1 = Depends(expensive_operation), data2 = Depends(expensive_operation)):
    # expensive_operation вызовется только один раз
    return {"data1": data1, "data2": data2}

Тестирование с зависимостями

from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient

app = FastAPI()

def get_db():
    return "real-database"

@app.get("/data")
async def get_data(db = Depends(get_db)):
    return {"db": db}

def test_get_data():
    def get_test_db():
        return "test-database"
    
    app.dependency_overrides[get_db] = get_test_db
    client = TestClient(app)
    response = client.get("/data")
    assert response.json() == {"db": "test-database"}
    app.dependency_overrides.clear()

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

• Переиспользуемость — одна зависимость используется в нескольких маршрутах
• Тестируемость — легко подменять зависимости тестовыми версиями
• Чистота кода — обработчики остаются малыми и сосредоточенными
• Управление ресурсами — автоматическое закрытие БД, файлов и т.д.
• Кэширование — результаты кэшируются в рамках запроса
• Валидация — зависимости могут содержать сложную валидацию параметров

Передача зависимостей в FastAPI — это мощный инструмент для написания масштабируемых и тестируемых API приложений.