Год с Dishka: какой он — модный DI-контейнер?

Привет, меня зовут Юрий, я уже год использую хайповый IoC‑контейнер dishka и хочу немного поделиться опытом эксплуатации. Мой проект — движок для городской ночной поисковой игры «Схватка» (вы могли играть в неё или в один из аналогов — «Энкаунтер» или «Дозоры»). У нас в городе очень маленькое ламповое комьюнити, для которого я и написал этот движок. По причине локальности (игроков — всего 50 человек), я не буду давать ссылки на что‑то, что можно потрогать, и прошу вас не искать. Я никогда не пытался оптимизировать этот код или готовить его к хабр‑эффекту. Однако проект полностью open source.

До dishka

Движок игры содержит два представления — REST API и Telegram‑bot. При разработке, для удобства, я запускаю их отдельно, а вот на проде запускаю и то и другое в одном процессе (для экономии RAM на сервере).

Таким образом у меня 3 main‑функции для запуска. Кроме того, есть ещё несколько вспомогательных утилит, которые я использую, чтобы выполнить какие‑то системные работы. Итого общее число main‑функций составляет уже 12 штук.

Всё бы ничего, но каждый из них требует какое‑то количество зависимостей для работы: пулы соединений к бд (с более высокоуровневыми DAO), клиенты доступа к файлам, соединения с Redis и так далее. При этом некоторые вещи требуют кроме создания ещё и финализации (закрытия соединений с БД например).

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

Да и в целом код состоит из месива синхронных и асинхронных контекстных менеджеров, try‑finally выражений, вложенных друг в друга несколько раз, чтобы в правильном порядке всё инициализировать и финализировать.
Один из не самых страшных вариантов:

async def main():
    paths = get_paths()

    setup_logging(paths)
    config = load_config(paths)
    retort = create_retort()
    file_storage = create_file_storage(config.file_storage_config)
    bot = create_bot(config)
    pool = create_pool(config.db)
    level_test_dao = create_level_test_dao()
    try:
        async with (
            pool() as session,
            create_redis(config.redis) as redis,
        ):
            dao = HolderDao(session, redis, level_test_dao)
            file_gateway = BotFileGateway(
                bot=bot,
                file_storage=file_storage,
                dao=dao.file_info,
                tech_chat_id=config.bot.log_chat,
            )
            bot_player = await dao.player.upsert_author_dummy()
            await dao.commit()
            ############ next is real main, all other was only dependencies setup
            await do_some_stuff(
                bot_player=bot_player,
                dao=dao,
                file_gateway=file_gateway,
                retort=retort,
                path=config.file_storage_config.path,
            )
    finally:
        await bot.session.close()
        close_all_sessions()

Я делал несколько попыток упростить эти main‑функции, но ни одна из них мне не нравилась, он всё равно был перегружен.

Вторая проблема, которая существовала постоянно — перегруженная aiogram middleware и FastAPI Depends, в обоих случаях примерно одинаковый код — прокинуть инициализированное в main‑функции и инициализировать связанное с запросом и тоже прокинуть в handler. Код до боли одинаковый и громоздкий и опять у нас контекстные менеджеры, вложенные в другие контекстные менеджеры.

Третья проблема — хэндлеры (роуты) принимают завивисимости в разном виде и перекладывают их в интеракторы юзкейсов (в более старой части — сервисный слой). При этом опять чуть‑чуть, но отличается способ работы в API и Bot‑представлении.

IoC-контейнеры

Периодически я поглядывал на разные IoC‑контейнеры, но каждый раз они мне не нравились: у одних не было асинхронной работы, у других всё работало на глобальных переменных (что очень пугает) Depends у FastAPI неплох, но работает только с FastAPI, у некоторых очень уж монструозное API. В итоге я каждый раз посмотрю часок доку, вздохну и продолжу страдать.

И вот однажды в одном из Telegram‑чатов я увидел новость, что вышел новый IoC‑контейнер Dishka. Посмотрев рассуждения автора библиотеки я обнаружил, что он при проектировании учёл все проблемы, которые меня беспокоили. Немного подождав других отзывов, решил попробовать втянуть к себе.

После добавлении Dishka

Удивительно (а может, кому‑то и неудивительно), но все эти проблемы полностью решены с помощью dishka.

main — это всё зависимости для области (scope) APP (в некоторых других di это называется singleton).

async def main():
    paths = common_get_paths()

    setup_logging(paths)
    dishka = make_async_container(
        *get_providers(paths),
    )
    try:
        config = await dishka.get(TgBotConfig)
        dao = await dishka.get(HolderDao)
        bot_player = await dao.player.upsert_author_dummy()
        await dao.commit()
        await do_some_stuff(
            bot_player=bot_player,
            dao=dao,
            file_gateway=await dishka.get(FileGateway),
            retort=await dishka.get(Retort),
            path=config.file_storage_config.path.parent / "scn",
        )
    finally:
        await dishka.close()

Мало того, что стало короче, так ещё и понятнее, а в придачу — никакого дублирования между main‑функциями.

Зависимости области (scope) REQUEST так же попали в dishka и нигде не дублируются.

Все хендлеры (роуты) стали выглядеть одинаково, всё можно переиспользовать как в различных фреймворках, так и без оных.

Правда, у меня не нашлось сил, чтобы всё переписать сразу, местами (на самом деле — довольно много где) ещё можно встретить легаси подход, однако dishka достаточно толерантен к любому сочетанию подходов и это не вызывает проблем. Я переписываю на новый подход с dishka, когда есть время и желание. Например, моя мидлварь для aiogram всё ещё выглядит как:

async def __call__(
    self,
    handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
    event: TelegramObject,
    data: MiddlewareData,
) -> Any:
    dishka = data["dishka_container"]
    file_storage = await dishka.get(FileStorage)  # type: ignore[type-abstract]
    data["config"] = await dishka.get(BotConfig)
    data["retort"] = await dishka.get(Retort)
    data["scheduler"] = await dishka.get(Scheduler)  # type: ignore[type-abstract]
    data["file_storage"] = file_storage
    holder_dao = await dishka.get(HolderDao)
    data["dao"] = holder_dao
    data["hint_parser"] = HintParser(
        dao=holder_dao.file_info,
        file_storage=file_storage,
        bot=data["bot"],
    )
    data["results_painter"] = ResultsPainter(
        data["bot"],
        holder_dao,
        data["config"].log_chat,
    )
    # ... and a lot of other deps
    result = await handler(event, data)
    return result

То есть я по‑прежнему перекладываю в middleware data многие зависимости, потому что в проекте очень много хендлеров и пока у меня нет сил все их исправить, но это и никак не мешает. Если я потрогаю хендлер, то, скорее всего, я его исправлю. Новый напишу сразу на новом подходе, а старые лежат, есть не просят, работают.

В сниппете выше можно заметить # type: ignore[type‑abstract] это артефакт прошлых версий dishka, где была проблема с выводом типов у mypy для протоколов и абстрактных классов. Сейчас эта проблема уже исправлена.

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

## api/routes/game.py
router = APIRouter(prefix="/games")


@router.get("/running/hints")
@inject
async def get_running_game_hints(
    user: FromDishka[dto.User],
    interactor: FromDishka[GamePlayReaderInteractor],
) -> responses.CurrentHintResponse:
    return responses.CurrentHintResponse.from_core(await interactor(user))

## infrastructure/di/interactors.py
class GamePlayProvider(Provider):
    scope = Scope.REQUEST

    # ... other factories

    @provide
    def game_play_reader(self, dao: HolderDao) -> GamePlayReader:
        return GamePlayReaderImpl(dao)
    
    game_play_reader_interactor = provide(GamePlayReaderInteractor)


## core/games/interactors.py
class GamePlayReaderInteractor:
    def __init__(self, reader: GamePlayReader):
        self.reader = reader

    async def __call__(self, user: dto.User) -> CurrentHints:
        ...
        # do actual stuff

Таким образом, мы в хендлере имеем ровно одну зависимость — Callable нашего интерактора — и какие‑то переменные, связанные с контекстом, в данном случае — пользователь, совершивший запрос.

В провайдере dishka всё настраивается тривиальным образом:

Reader у меня настраивается через функцию, чтобы было видно, какую имплементацию подставляют под интерфейс (хотя dishka в новых версиях уже позволяет и такие функции не писать, всё можно описывать декларативнее, но в некоторых сложных случаях функции всё ещё нужно писать), в то же время интерактор создаётся автоматически, без функции, по анализу конструктора.

Сам интерактор тоже имеет очень удобное разделение. В __init__ закладываются все зависимости с помощью dishka, а __call__ же вызывается с контекстом из роута (хендлера).

Тестирование с dishka

С переменным успехом я стараюсь писать тесты (покрытие на момент написания статьи — 69%). Часто это просто e2e тесты или функциональные тесты на интеракторы юзкейсов. Иногда в местах особо сложной бизнес‑логики, где я осилил её выделение в синхронные функции и методы без IO, есть ещё и unit‑тесты.

Естественно, в разговоре о dishka unit‑тесты не имеют никакого значения, там мы тестируем конкретные вещи, передавая всё явным образом.

В функциональных тестах dishka, скорее всего, тоже не нужен, поскольку pytest.fixture отлично справляются (ведь по факту pytest тоже является ioc‑контейнером и очень даже неплохим), однако иногда я тут всё же использую dishka, если зависимости имеют сложную логику инициализации и финализации, и мокировать их не требуется. Тогда я просто получаю в тесте контейнер dishka через fixture и достаю нужную зависимость.

В случае же с e2e тестами dishka необходим. Контейнер обязательно надо подсунуть в приложение, чтобы оно использовало моки, а не оригинальные зависимости (например, чтобы не ходило во внешнюю сеть или использовало тестовую БД) Мой контейнер для тестов собирается примерно так:

@pytest_asyncio.fixture(scope="session")
async def dishka():
    mock_provider = Provider(scope=Scope.APP)
    mock_provider.provide(GameLogWriterMock, provides=GameLogWriter)
    mock_provider.provide(UserGetterMock, provides=UserGetter)
    mock_provider.provide(SchedulerMock, provides=Scheduler)
    mock_provider.provide(ClockMock)
    container = make_async_container(
        ConfigProvider("SHVATKA_TEST_PATH"),
        TestDbProvider(),
        #...
        GamePlayProvider(),
        GameToolsProvider(),
        mock_provider,
    )
    yield container
    await container.close()

@pytest_asyncio.fixture
async def dishka_request(dishka: AsyncContainer):
    async with dishka() as request_container:
        yield request_container

Выше показано, что у меня есть две fixture с контейнером: одна — в скоупе APP, другая — в скоупе REQUEST. В разных целях могут использоваться один или другой. В e2e тесте мы, как принято в тестах для FastAPI, делаем примерно следующее:

@pytest.fixture(scope="session")
def app(dishka: AsyncContainer, api_config: ApiConfig):
    app = create_app(api_config)
    setup_dishka(dishka, app)
    return app


@pytest_asyncio.fixture(scope="session")
async def client(app: FastAPI):
    async with (
        AsyncClient(app=app, base_url="http://test") as ac,
    ):
        yield ac

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

@pytest.mark.asyncio
async def test_game_file(
    finished_game: dto.FullGame,
    client: AsyncClient,
    auth: AuthProperties,
    user: dto.User,
):
    token = auth.create_user_token(user)
    resp = await client.get(
        f"/games/{finished_game.id}/files/{GUID}",
        cookies={"Authorization": "Bearer " + token.access_token},
    )
    assert resp.is_success
    assert resp.read() == b"123"

Выводы

За год работы с dishka в моём не самом маленьком проекте (30к строк, 573 файла) у меня образовалось 52 фабрики в 21 провайдере, и мне по‑прежнему всё очень нравится, я до сих пор не встретил ни одной проблемы кроме того случая с mypy, которую уже исправили. В качестве бонуса прикреплю сюда сгенерированную с помощью dishka картинку со связями моих зависимостей https://gist.github.com/user-attachments/assets/72813759-3303-4c04-9671-befd49c0f8a8

• Репо dishka https://github.com/reagento/dishka

• Документация dishka https://dishka.readthedocs.io/en/stable/

• Репо моего проекта https://github.com/bomzheg/Shvatka

Под редакцией Антона Швецова