В свое время 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, а это позволяет нам
Заменить Redis на Memcached или даже хранение в памяти при необходимости
Поместить в качестве 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. Но, кажется, создатель отказался от его сопровождения (да и библиотека имеет множество минусов, которые нам тоже не нравятся).
А что нам остается?
Все как-то не то. Слабое распространение, скудный функционал, нестабильный API.
В общем, в ходе продолжительных баталий дискуссий опытный разработчик Андрей Тихонов (автор канала Советы разработчикам, администратор ru-python, fastapi-ru и прочих крупных TG-групп) решил создать собственное решение - dishka!
Полное сравнение с другими библиотеками вы можете найти в документации библиотеки.
Но я пределагаю сначала взглянуть, как он работает, а потом уже сравнить с FastAPI Depends.
Использование dishka
Концепция проста и незамысловата:
Мы пишем "провайдеры" - классы, которые содержат в себе фабрики зависимостей
Затем объединяем их в "контейнер", откуда они уже будут доставляться в конечные функции
Используем интеграции с фреймворками для бесшовного встраивания контейнера в ваше приложение
Пишем "провайдер"
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 (как в моем примере) и/или lifespanrequest.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
подписаться на мой телеграм канал, если вам интересен подобный материал
