Привет, Хабр, меня зовут Юрий, я уже год использую хайповый IoC‑контейнер dishka в python-проекте и хочу немного поделиться опытом эксплуатации. Мой проект — движок для городской ночной поисковой игры «Схватка» (вы могли играть в неё или в один из аналогов — «Энкаунтер» или «Дозоры»). У нас в городе очень маленькое ламповое комьюнити, для которого я и написал этот движок. По причине локальности (игроков — всего 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
Под редакцией Антона Швецова
ссылка на оригинал статьи https://habr.com/ru/articles/894286/
Добавить комментарий