FastAPI и Dependency Injection: правда или вымысел?

В свое время FastAPI прогремел как гром среди ясного неба - тут тебе и минималистичный API аля-Flask (все устали от Django, диктующего свои правила), и OpenAPI документация из коробки, и удобное тестирование, и хайповая асинхронность. Буквально все, что нужно для свободы творчества, и никаких ограничений! Да еще и Depends завезли! В тот момент это был культурный шок - Dependency Injection в Python? Разве это не что-то из Java?

FastAPI показал, что DI - это паттерн, упрощающий разработку вне зависимости от языка программирования. Теперь DI как фича является практически неотъемлемым элементом любого нового Python-фреймворка (Litestar/Blacksheep/FastStream/etc), ведь людям это нужно. Все хотят "как в FastAPI".

Но дьявол кроется в деталях. А вы уверены, что те самые Depends == Dependency Injection? Уверены, что пишете код на FastAPI правильно?

Что Tiangolo (создатель FastAPI) прививает вам "лучшие практики"?

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

Что такое DI и зачем он нам нужен?

Dependency Injection - это паттерн, сильно помогающий следовать принципу Инверсии зависимостей (DIP - Dependency Inversion Principle) из soliD.

DIP заключается в том, что наша бизнес-логика не должна зависеть от деталей реализации (базы данных, протокола взаимодействия, конкретных библиотек). Вместо этого она должна запрашивать абстрактные интерфейсы, декларирующие методы, которые ей необходимы. Эти "абстрактные интерфейсы" находятся в ядре вашей системы, т.к. жизненно необходимы для ее функционирования.

А вот с помощью паттерна Dependency Injection "реальные" имплементации этих интерфейсов (которые знают про конкретные базы данных и тд) будут доставляться в вашу логику извне при инициализации проекта (тот самый Injection).

Т.е. вместо подобного кода:

class TokenChecker:
    def __init__(self) -> None:
        self.storage = Redis()

    def check_token(self, token: str) -> bool:
        return self.storage.get(token) is not None

checker = TokenChecker()

Мы должны писать нечто такое:

from typing import Protocol

# находится в БЛ, так как нужен для ее функционирования
class Storage(Protocol):
    def get(self, token: str) -> str | None:
        ...

class TokenChecker:
    def __init__(self, storage: Storage) -> None:
        self.storage = storage

    def check_token(self, token: str) -> bool:
        return self.storage.get(token) is not None

real_storage = Redis() # объект Redis подходит под интерфейс Storage
checker = TokenChecker(real_storage)

Кода стало больше, но зачем? - Теперь TokenChecker больше не знает о том, что работает с Redis, а это позволяет нам

1. Заменить Redis на Memcached или даже хранение в памяти при необходимости

2. Поместить в качестве Storage mock-объект в тестах удобным и понятным способом

Изначальная мотивация действительно пришла из Java и других компилируемых языков. Смысл в том, что внешний слой подвергается изменениям часто, а вот внутренний - редко. Если мы зависим от внешнего слоя в нашей бизнес-логике (банально делаем импорты оттуда), то при повторной компиляции проекта эти модули также придется перекомпилировать, хотя изменений в них не произошло (изменения были в модулях, от которых они зависят). Неконтролируемые зависимости приводят к тому, что весь проект пересобирается при изменении любой строки в любом файле и тем самым многочасовым "код компилируется".

Однако, DI - это хорошая практика, которая приносит ощутимую пользу в любых языках.

Иногда вы можете встретить еще и формулировку Inversion of Control (IoC), что суть - о том же самом. Когда мы следуем подходу Dependency Injection, у нас образуется отдельная группа функций и классов, выполняющих только одну задачу - создание других объектов.
В сложном приложении такой компонент может содержать большое количество функций, контролировать как создание, так и корректную очистку объектов и, что самое главное - их взаимосвязь. Для упрощения работы с такими фабриками придумали отдельный тип библиотек - IoC-контейнеры (DI-фреймворки).

DI в FastAPI по Tiangolo

Одна из основных фич FastAPI - его Depends, которая как раз позиционируется как реализация Dependency Injection принципа. Давайте посмотрим, как Tiangolo предлагает ее использовать:

from typing import Annotated

from fastapi import Depends

async def common_parameters(
    q: str | None = None,
    skip: int = 0,
    limit: int = 100,
):
    return { "q": q, "skip": skip, "limit": limit }

@app.get("/items")
async def read_items(
    commons: Annotated[dict, Depends(common_parameters)],
):
    return commons

@app.get("/users")
async def read_users(
    commons: Annotated[dict, Depends(common_parameters)],
):
    return commons

В данном примере FastAPI распознает функцию common_parameters как зависимость, т.к. она была передана в Depends. При поступлении запроса на read_users обработчик, FastAPI вызовет все "зависимости" данного метода, а затем передаст результаты их выполнения в качестве аргументов основной функции. Подробнее о том, как это работает, можно прочитать в документации FastAPI

"Так вы можете переиспользовать логику между разными эндпоинтами" - вот как аргументирует использование Depends Tiangolo. Однако, это не Dependency Injection.

Просто давайте взглянем на следующий код:

from typing import Annotated

from fastapi import Request

async def common_parameters(
    q: str | None = None,
    skip: int = 0,
    limit: int = 100,
):
    return { "q": q, "skip": skip, "limit": limit }

@app.get("/items")
async def read_items(request: Request):
    commons = await common_parameters(**request.query_params)
    return commons

@app.get("/users")
async def read_users(request: Request):
    commons = await common_parameters(**request.query_params)
    return commons

Разве это не то же самое "переиспользование логики", с которым нам хочет помочь Tiangolo? Кажется, его помощь - это просто еще один слой синтаксического сахара (не бесплатного, конечно).

Однако, Dependency Injection тут все-таки есть, т.к. есть возможность заменить зависимость через механизм dependency-overrides

async def override_dependency(q: str | None = None):
    return {"q": q, "skip": 5, "limit": 10}

app.dependency_overrides[common_parameters] = override_dependency

В данном случае мы подменяем все "зависимости" вида Depends(common_parameters) на Depends(override_dependency) по всему проекту. Т.е., когда запрос придет в обработчик, вместо оригинальной функции common_parameters будет вызвана override_dependency вне зависимости от сигнатуры самого обработчика.

В варианте с прямым использованием функции это невозможно.

Правда, механизм позиционируется "для тестов" и все еще не помогает соблюсти DIP - мы подменяем зависимость от реализации на другую зависимость от реализации. Что может только путать людей, работающих с кодовой базой.

Но не все потеряно и мы можем доработать Depends так, чтобы это был настоящий DI с соблюдением DIP.

"Настоящий" DI в FastAPI

Не претендую на авторство данного подхода, но готов принять все шишки за его использование, т.к. не нашел способа сделать DI лучше.

Так вот: помним, что в DI нам нужно завязывать на абстракцию, а реализацию Inject'ить?

В FastAPI МОЖНО реализовать Dependency Injection с соблюдением DIP. Но не совсем тем способом, которым планировал Tiangolo.

В FastAPI у нас есть глобальный словарь app.dependency_overrides, который предлагается использовать для "тестирования зависимостей" (в документации). Однако, по всем внешним признакам - это контейнер зависимостей. И мы можем его использовать как раз по прямому назначению IoC контейнера - Inject'ить зависимости.

Давайте разбираться по порядку.

Вводим абстракцию

Давайте представим, что нам нужно идентифицировать пользователя по наличию токена в кеше? Код будет несколько упрощен относительно реального, но смысл от этого становится только яснее.

Для начала введем интерфейс объекта, с помощью которого мы как раз будем валидировать токен:

from typing import Protocol

class TokenRepo(Protocol):
    async def get_user_by_token(self, token: str) -> str | None:
        ...

    async def set_user_token(self, token: str, username: str) -> None:
        ...

Зависим от абстракции

Теперь нам нужно "завязаться" на эту абстракцию в нашем эндпоинте:

from typing import Annotated
from fastapi import FastAPI, Depends

app = FastAPI()

@app.get("/{token}")
async def get_user_by_token(
    token: str,
    token_repo: Annotated[TokenRepo, Depends()],  # "запрашиваем" абстракцию
) -> str | None:
    return await token_repo.get_user_by_token(token)

Пишем реализацию

Нам остается только реализовать где-то заданный интерфейс и поместить эту реализацию в наш контейнер зависимостей (откуда она попадет в исполняемый код вместо абстракции).

Реализация протокола для работы с Redis:

from redis.asyncio import Redis

class RedisTokenRepo(TokenRepo):
    def __init__(
        self,
        redis: Redis,
        expiration: str,
    ) -> None:
        self.redis = redis
        self.token_expiration = expiration

    async def get_user_by_token(self, token: str) -> str | None:
        if username := await self.redis.get(token):
            return username.decode()

    async def set_user_token(self, token: str, username: str) -> None:
        await self.redis.set(
            name=token,
            value=username,
            ex=self.token_expiration,
        )

Используем реализацию вместо абстракции

Ну и "помещаем" нашу реализацию в FastAPI IoC Container:

def setup_ioc_container(
    app: FastAPI,
) -> FastAPI:
    settings_object = {  # mock настроек
        "redis_url": "redis://localhost:6379",
        "token_expiration": 300,
    }

    redis_repo = RedisTokenRepo(
        redis=Redis.from_url(settings_object["redis_url"]),
        expiration=settings_object["token_expiration"],
    )

    app.dependency_overrides.update({
        TokenRepo: lambda: redis_repo,
    })

    return app

Реальной зависимостью в нашем случае является lambda: redis_repo. Именно эта функция будет вызываться при каждом запросе с Annotated[TokenRepo, Depends()] зависимостью.

Мы реализовали ее через lambda для того, чтобы избежать вызова конструктора RedisTokenRepo на каждый вызов, а сделать этот объект "синглтоном".

Так выглядит DI в FastAPI "здорового человека". Но не совсем.

Боремся с FastAPI

К сожалению, Tiangolo не планировал использование Depends таким образом. Он не хочет, чтобы мы зависели от "абстракции". Поэтому в нашу OpenAPI схему просочилось что-то странное (args, kwargs?):

Это происходит потому что FastAPI парсит сигнатуру "зависимости", которую мы запрашиваем (Annotated[TokenRepo, Depends()]), а именно - __init__ метод класса.

class TokenRepo(Protocol):
     # init класса Protocol по умолчанию содержит args, kwargs
     def __init__(self, *args, **kwargs): ...

Вот FastAPI и нашел "лишние" аргументы и нарисовал их в сигнатуре.

Для того, чтобы от этого избавить нужно "спрятать" от FastAPI сигнатуру исходной "абстракции". (Можно еще отнаследоваться от abc.ABC вместо typing.Protocol, но это уже "протекание" деталей FastAPI в наши абстракции, чего мы не хотим)

Сделать это можно следующим образом:

from typing import Callable, Any

class Stub:
    def __init__(self, dependency: Callable[..., Any]) -> None:
        """Сохраняем нашу абстракцию."""
        self._dependency = dependency

    def __call__(self) -> None:
        """Выкинем ошибку, если забыли подменить реализацию при старте приложения."""
        raise NotImplementedError(f"You forgot to register `{self._dependency}` implementation.")

    def __hash__(self) -> int:
        """Обманываем app.dependency_overrides, чтобы он считал Stub реальной зависимостью"""
        return hash(self._dependency)

    def __eq__(self, __value: object) -> bool:
        """Обманываем app.dependency_overrides, чтобы он считал Stub реальной зависимостью"""
        if isinstance(__value, Stub):
            return self._dependency == __value._dependency
        else:
            return self._dependency == __value

Теперь мы должны "запрашивать" зависимость следующим образом:

@app.get("/{token}")
async def get_user_by_token(
    token: str,
    token_repo: Annotated[TokenRepo, Depends(Stub(TokenRepo))]
): ...

Уже не так "сахарно", зато в схему ничего не течет.

Резюме

Dependency Injection в FastAPI возможен, однако:

• требует дополнительных приседаний, чтобы ничего не утекало в схему

• требует дополнительных приседаний для реализации Application-level зависимостей (объктов, которые создаются 1 раз при старте приложения)

• требует дополнительных приседаний для регистрации зависимостей

Альтернатива?

Допустим, DI в FastAPI нам не сильно нравится (а он нам не нравится) и мы хотим взять стороннюю библиотеку. Наверное, первое, что приходит на ум - это Dependency Injector. Но, кажется, создатель отказался от его сопровождения (да и библиотека имеет множество минусов, которые нам тоже не нравятся).

А что нам остается?

• rodi ?

• punq ?

• di ?

• FastDepends ?

Все как-то не то. Слабое распространение, скудный функционал, нестабильный API.

В общем, в ходе продолжительных баталий дискуссий опытный разработчик Андрей Тихонов (автор канала Советы разработчикам, администратор ru-python, fastapi-ru и прочих крупных TG-групп) решил создать собственное решение - dishka!

Полное сравнение с другими библиотеками вы можете найти в документации библиотеки.

Но я пределагаю сначала взглянуть, как он работает, а потом уже сравнить с FastAPI Depends.

Использование dishka

Концепция проста и незамысловата:

1. Мы пишем "провайдеры" - классы, которые содержат в себе фабрики зависимостей

2. Затем объединяем их в "контейнер", откуда они уже будут доставляться в конечные функции

3. Используем интеграции с фреймворками для бесшовного встраивания контейнера в ваше приложение

Пишем "провайдер"

from dataclasses import dataclass

from dishka import Provider, Scope, provide, from_context

@dataclass
class Settings:
    redis_url: str
    token_expiration: int

class RepoProvider(Provider):
    # говорим, что объект типа Settings будем помещаться в контейнер пользователем
    settings = from_context(provides=Settings, scope=Scope.APP)

    @provide(scope=Scope.APP)  # зависимость уровня приложения (синглтон)
    def get_redis_token_repo(
        self,
        settings: Settings,  # "запрашиваем" другую зависимость
    ) -> TokenRepo:
        return RedisTokenRepo(
            redis=Redis.from_url(settings.redis_url),
            expiration=settings.token_expiration,
        )

Затем мы должны собрать из провайдера (у нас он один) контейнер

container = make_async_container(
    RepoProvider(),
    context={  # помещаем Settings в контейнер вручную
        Settings: Settings(
            redis_url="redis://localhost:6379",
            token_expiration=300,
        ),
    },
)

И наконец - используем этот контейнер в нашем FastAPI приложении!

from fastapi import APIRouter, FastAPI
from dishka.integrations.fastapi import FromDishka, DishkaRoute, setup_dishka

router = APIRouter(route_class=DishkaRoute)  # используем специальный route_class

@router.get("/{token}")
async def get_user_by_token(
    token: str,
    token_repo: FromDishka[TokenRepo],  # используем вместо Depends
) -> str | None:
    return await token_repo.get_user_by_token(token)

app = FastAPI()
app.include_router(router)
setup_dishka(container, app)

Как видите, приседаний стало меньше, сайд-эффектов - тоже, а контейнер можно переиспользовать и для других фреймворков/библиотек, если они запущены в том же рантайме (например, FastStream).

Выводы по dishka

По моему субъективному мнению dishka значительно комфортнее для реализации DI принципа в FastAPI проекте (относительно нативного Depends) по следуюшим причинам:

• Имеет четкое разделение на Application-level (синглтоны в рамках приложения) и Request-level (создаются на каждый запрос) зависимости. Нативный Depends работает только для Request зависимостей, а Application-level (самые частые) приходится изобретать самостоятельно вокруг main (как в моем примере) и/или lifespan request.state.* (как советует Starlette). Также dishka поддерживает и другие Scope'ы, в т.ч. и кастомные, что позволяет использовать его в совершенно разных кейсах.

• Финализация Application-level зависимостей. В FastAPI отдельной головной болью стоит вопрос о том, как их финализировать, а для асинхронных зависимостей - еще и инициализировать (асинхронный main? извращения с lifespan?). Dishka поддерживает как асинхронные фабрики зависимостей, так и фабрики с yield, так что обе проблемы для него просто не существуют.

• Помогает организовать логику управления графом зависимостей в одном месте, не размазывая ее по разным функциям и частям приложения (а также избавляет от различных служебных функций-оберток, необходимых для победы над FastAPI)

• Позволяет переиспользовать контейнер зависимостей в рамках всего приложения (и других фреймворков/библиотек), а не только handler'ах FastAPI (аккуратнее с этим). Также, вы без труда сможете мигрировать на другой веб-фреймворк без переписывания логики DI. HTTP-фреймворк в таком случае остается только на транспортном уровне, где мы и хотим его видеть.

• Работает несколько быстрее стандартного Depends

Однако, у использования dishka есть и свои минусы

• Придется потратить 20 минут на изучение новой библиотеки

• +1 зависимость (и библиотека в вашем портфолио)

• В уже созданном контейнере нельзя переопределить зависимости, поэтому организация main должна учитывать, что в тестах вам потребуется использовать другие контейнеры под разные сценарии

• Возможность разделения фабрик зависимостей на логические группы (по разным провайдерам) может вскружить голову и вы сделаете хуже, чем было до dishka. Поэтому рекомендую начинать с 1го провайдера на приложение, а там - как пойдет.

• Нет возможности учитывать зависимости при генерации OpenAPI

Выводы

FastAPI сделал неоценимый вклад в Python-экосистему. Он виртуозно объединил в себе лучшие фичи уже существующих решений и показал, каким должен быть современный инструмент. Однако, его документация, к сожалению, может вводить пользователей в заблуждение относительно тех или иных понятий и подходов, а детали реализации накладывают на пользователя свои ограничения.

В данной статье мы рассмотрели самый популярный и "правильный" подход к реализации принципа внедрения зависимостей в рамках FastAPI приложения, а также познакомились с dishka - великолепной библиотекой, которая позволяет реализовать DI в рамках любого приложения (в т.ч. и FastAPI).

Лично я рекомендую вам как минимум обратить внимение на эту библиотеку, а еще:

• поддержать ее автора, поставив звезду на GitHub

• вступить в телеграм чат, где вы можете пообщаться с ее создателем лично

• прочитать статью про использование dishka с Litestar и FastStream

• посмотреть доклад автора dishka на Podlodka Python Crew

• подписаться на мой телеграм канал, если вам интересен подобный материал