структура текста:
1. DI контейнер, разбираем Dependency-Injector
2. Что такое фабрика
2.1 Создаём контейнер для метода класса
2.1.1 Решаем супер-простую задачу. База данных, API получение профиля пользователя
2.1.2 Пишем зависимости
2.1.3 Пишем обработчик с этими зависимостями
2.2 Как происходит резолвинг зависимостей
2.2.1 Получить
2.2.2 Resource-> Singletone (умеет в асинхронность)
2.2.3 Factory, который умеет только в синхронность
2.2.4 Подставновка зависимостей (исходный код)
2.2 Шаблоны фабрик и вложение зависимостей
2.2.1 Пишем зависимости с синхронным и асинхронным кодом
2.2.2 Нарисуем рядышком фотогрфию с дереров зависимостей
2.2.3 Показываем код инъектирования зависимостей
2.2.4 Показываем как подстановка зависимости не до конца реализуется в контейнере
2.2.5 Показывем, как автоматически это поправить
2.2.6 Показывем, как по названию фабрики не работает её поиск и реализация в дереве зависимостей
2.3 Микровывод, о том, что билиотека хорошая, но надо понимать и ограничивать тем, что мы используем от сервиса
2.3.1
2.4
Эта статья демонстрирует различные варианты работы с библиотекой Dependecy Injector на python и содержит некоторые комментарии и ограничения для работы с подходами как при тестировании, так и при первом старте проекта. Для более плавного знаковства рекомендуется прочитать [документацию](https://pypi.org/project/dependency-injector/).
В примерах будем использовать код на FastAPI.
Начнём с решения небольшой задачки. Реализуем API, подключим и декларативного описания зависимостей в
Рассмотрим три варианта подключения готового контейнера
(1)
1. Пишем функцию которая асинхронно подключается к базе данных и передаёт объект с подключением
2. Используем Resource для того, чтобы функция возвратила объект подключения к базе
3.
(2)
## Начинается статья с этого момента
В этой статье мы рассмотрим инструмент управления зависимости любимой (нет) библиотеки - Dependency Injector в процессе разработки на Python. Dependecy Injector - это мощный инструмент, который позволяет упростить управление зависимостями в проекте, достаточно большой и предоставляет класальные возможности по управлению зависимости в проекте. В примерах будут использованы код на FastAPI, поскольку в продакшене новые сервисы начинают чаше всего пилиться на нём (типа стандарт).
Начнём с простой задачки. Создадим маленький сервис, который реализуйте API для получения профиля пользователя по его внутреннему идентификатору.
Напишем [UseCase](https://www.cosmicpython.com/book/chapter_04_service_layer.html) и метод [репозитория](https://www.cosmicpython.com/book/chapter_02_repository.html), которые могут использовать зависимости друг от друга. Их код будет выглядеть примерно так:
```python
import logging
from abc import ABC, abstractmethod
from pydantic import UUID4
from models import Profile
logger = logging.getLogger(__name__)
class UserRepositoryBase(ABC):
@abstractmethod
async def get_by_id(self, entry_id: UUID4) -> Profile:
...
class GetProfileUseCase:
def __init__(
self,
user_repository: UserRepositoryBase,
):
self.user_repository = user_repository
async def execute(
self,
profile_id: UUID4
) -> Profile:
logger.info(f"Получение профиля пользователя по его идентификатору profile_id={str(profile_id)}")
return await self.user_repository.get_by_id(entry_id=profile_id)
```
Обыычно выделяют core уровень для хранения всех UseCase'ов и интерфейсов (абстрактных классов) и application уровень для самих реализаций. Но очень чистая архитектура для **действительно** небольшого сервиса может быть избыточной
Затем, нам необходимо описать декларативно зависимости нашего проекта. Например можно описывать их следующим образом:
```python
from dependency_injector import containers, providers
from databases import Database
class BaseRepositoriesContainer(containers.DeclarativeContainer):
database = providers.Dependency(instance_of=Database)
# implement UserRepositoryBase
user_repo = providers.Factory(UserRepository, database=database)
class UseCasesContainer(containers.DeclarativeContainer):
repositories: BaseRepositoriesContainer = providers.DependenciesContainer()
get_profile_uc = providers.Factory(
GetProfileUseCase,
user_repository=repositories.user_repo
)
class ApplicationDI(containers.DeclarativeContainer):
database = providers.Resource(connect_db)
repositories = providers.Container(BaseRepositoriesContainer, database=database)
use_cases = providers.Container(
UseCasesContainer,
repositories=repositories
)
```
В данном случае функция connect_db асинхронная и возвращает объект подключения к БД для всего приложения. [Resource]() является провайдером, который **единожды** выполняет переданную в него функцию и возвращает объект в момент вызова реализации зависимостей.
Обработчик приложения может выглядит следующим образом:
* ключевое слово [inject]() будет создавать обертку над функцией, для того, чтобы реализовать переданные [Provide]() объекты в момент вызова.
* Ключевое слово Provide указывает на путь зависимости созданного контейнера
```python
from fastapi import APIRouter
from pydantic import UUID4
route = APIRouter(prefix="/api/v1/profiles")
@route.get(path="/get/{profile_id}", response_model=Profile)
@inject
async def profile_get_by_id(
profile_id: UUID4,
uc=Depends(Provide[ApplicationDI.use_cases.get_profile_uc]),
) -> Profile:
return await uc.execute(profile_id=profile_id)
```
Инициализируем контейнер
```python
def init_di() -> ApplicationDI:
logger.info("Запуск DI-контейнера")
app_di = ApplicationDI()
return app_di
def get_app():
_app = FastAPI(title="Backend")
app_di = init_di()
_app.state.container = app_di
return _app
app = get_app()
```
При старте проекта мы можем столкнуться с проблемой [recursion limit]()
Первый костыль, с которым мы познакомимся, заключается в инициализации контейнера. При попытке стартовать контейнер с более чем одной зависимостью мы можем упасть с ошибкой recursion limit.
```
RecursionError: maximum recursion depth exceeded in comparison
```
Добавляем магические строки, чтобы у нас всё заработало:
Вызываем метод API и падаем с ошибкой в логах
```
Provide has not execute attrubute
```
Для работы с Dependency Injector необходимо связывать те функции в модулях с контейнером для иъецирования (внедрения) зависимостей внутрь функции. Для этого напишем функцию:
```python
def wire_di_modules(app_di):
from infrastructure.web.views import profile_view
app_di.wire(
modules=[
profile_view,
],
)
logger.info("Применение DI-контейнера к приложению")
return app_di
```
Теперь реализация приложения FastAPI должна выполняться следующим образом:
```python
def get_app():
_app = FastAPI(title="Backend")
app_di = wire_di_modules(init_di())
_app.state.container = app_di
return _app
app = get_app()
```
Расскажу про провайдеры, которыми мы пользовались в проекте:
1. Factory
1. Провайдер который создаст класс переданный в него с теми параметрами, которые указаны внутри. На каждый вызов функции будет создан экземпляр класса и передат (инджектирован) в функцию. То есть каждый UseCase который мы пишем в обработчике API работает независимо, и, если нам неоюходимо менять состояние UseCase'а для своих нужд, оно сохранится только в одном экземпляре этого класса
2. Singletone
1. Провайдер который создаст класс переданный в него с теми параметрами, которые указаны внутри, но только один раз на проект. Полезно, когда класс не нуждается в дополнительном стейте, это могут быть API-клиенты, работающие с внешними сессиями и т.д
3. Resource
1. Провайдер который принимает функцию (синхронную или асинхронную), и возвращает объект, который будет являться Signleton'ом
4. FactoryAggregate
1. Для использования шаблонного метода, необходимо под один интерфейс в зависимости от имени реализации подбрасывать (реализовывать) одну из фабрик. Чтобы проще этим можно было управлять, используется агрегация фабрик. Рассмотрим пример получения профиля из внешней системы и из нашей базы данных
1. Пишем интерфейс
2. Подключаем две реализации
3. Агрегируем обе фабрики в этот интерфейс
Решим ещё одну задачу. Теперь мы хотим в зависимости от параметра запроса получить профиль пользователя из разных источников (подбрасывать в рантайме разную реализацию)
```plantuml
@startuml
skinparam linetype ortho
class "UseCase" as H
class "FactoryExternal" as F1
class "FactoryDatabase" as F2
class "FactoryAggregate" as FA
class "FactoryUseCase" as FC
class "UserExternalRepository" as C2
class "UserDatabaseRepository" as C3
FC <-- FA
F1 <-- C2
F2 <-- C3
FA <-- F1
FA <-- F2
FC <-- H
@enduml
```
```python
class BaseRepositoriesContainer(containers.DeclarativeContainer):
database = providers.Dependency(instance_of=Database)
user_database_repo = providers.Factory(UserDatabaseRepository, database=database)
user_external_repo = providers.Factory(UserExternalRepository, ...)
user_repo = providers.FactoryAggregate(
database=user_database_repo,
external=user_external_repo,
)
class UseCasesContainer(containers.DeclarativeContainer):
repositories: BaseRepositoriesContainer = providers.DependenciesContainer()
get_profile_uc = providers.Factory(
GetProfileUseCase,
user_repository=repositories.user_repo # <-- FactoryAggregate
)
class ApplicationDI(containers.DeclarativeContainer):
database = providers.Resource(connect_db)
repositories = providers.Container(BaseRepositoriesContainer, database=database)
use_cases = providers.Container(
UseCasesContainer,
repositories=repositories
)
```
Для простоты, мы используем в UseCase стратегию подстановки зависимости следующим образом:
```python
class GetProfileUseCase:
def __init__(
self,
user_repository: Callable[[str], UserRepositoryBase],
):
self.user_repository = user_repository
async def execute(
self,
profile_id: UUID4,
source: str,
) -> Profile:
logger.info(f"Получение профиля пользователя по его идентификатору profile_id={str(profile_id)}")
factory = self.user_repository(source)
return await factory.get_by_id(entry_id=profile_id)
```
Тогда обработчик становится не сильно больше, но решает нашу задачу
```python
@route.get(path="/get/{profile_id}", response_model=Profile)
@inject
async def profile_get_by_id(
profile_id: UUID4,
source: str,
uc=Depends(Provide[ApplicationDI.use_cases.get_profile_uc]),
) -> Profile:
return await uc.execute(profile_id=profile_id, source=source)
```
*Примечание. Частое использование шаблонных методов может привести к переусложнению бизнес логики в зависимости от ряда параметров передаваемых к реализации UseCase'ов*
*Примечание 2. Вы можете использовать Generic в типах данных в FactoryAggregate, например*
Запускаем код и пытаемся несколько раз использовать ручку.
Иногда падаем в рантайме с ошибкой для `source='external'`:
```python
AttributeError: '_asyncio.Future' object has no attribute 'get_by_id'
```
Заменяем строчку на `factory = await self.user_repository(source)` и всё равно падаем с ошибкой, но уже другой для `source='database'`:
```python
TypeError: object UserDatabaseRepository can't be used in 'await' expression
```
Это происходит, поскольку после FactoryAggregate всегда отдаёт синхронную реализацию объекта с зависимостями. Поскольку две фабрики зависят от асинхронной реализации зависимости и от синхронных, то Dependency Injector для реализации может передать на выход объект в одном случае и объект Future в другом.
Это можно решить миксиной для всех фабрик, которые мы используем. Необходим создать асинхронную обёртку, в которой:
1. Если фабрика возвращает объект Coroutine или Future, то после await операции - необходимо возвращать результат самой фабрики.
2. А если пришёл объект фабрики использующейся в шаблонах, то await для фабрики тоже отработал и вернул сам себя.
> В новых проектах мы не используем объекты Resource для подключения к внешним сервисам, а регистрируем их при старте проекта контекстным менеджером
> В новых проектах мы используем чаще вместо шаблонного метода и FactoryAggregate паттерны типа Стратегия (Strategy) или Фасад (Facade) для подключения различной реализации по переданным параметрам
## Дальше живут драконы ещё не доделано
```python
@inject
async def profile_get_by_id(
profile_id: UUID4,
uc=Depends(Provide[ApplicationDI.use_cases.get_profile_uc]),
) -> Profile:
return await uc.execute(profile_id=profile_id)
inspect.signature(profile_get_by_id)
inspect.getfullargspec(profile_get_by_id)
```
```python
(profile_id: pydantic.types.UUID4,
use_case=<Provide object at 0x1198cf7c0>
) -> Profile
FullArgSpec(
args=[],
varargs='args',
varkw='kwargs',
defaults=None,
kwonlyargs=[],
kwonlydefaults=None,
annotations={'return': <class 'Profile'>}
)
```
В момент вызова каждый provide знает какие зависимости у него есть и пытается их реализовать, записывая реализованный объект в стек. Так же, функция Depends из fastapi отменяется при инджектировании:
```python
def _is_fastapi_depends(param: Any) -> bool:
return fastapi and isinstance(
param,
fastapi.params.Depends
)
# Получение чистых зависимостей
# использующиеся внутри fn
def _fetch_reference_injections(
fn: Callable[..., Any],
) -> Tuple[Dict[str, Any], Dict[str, Any]]:
signature = inspect.signature(fn)
for parameter_name, parameter in signature.parameters.items():
...
marker = parameter.default
if _is_fastapi_depends(marker):
marker = marker.dependency
# Создание обёртки функции,
# которая при вызове реализует и подставит зависимости
def _get_patched(
fn: F,
reference_injections: Dict[Any, Any],
reference_closing: Dict[Any, Any],
) -> F:
patched_object = PatchedCallable(
original=fn,
reference_injections=reference_injections,
reference_closing=reference_closing,
)
if inspect.iscoroutinefunction(fn):
patched = _get_async_patched(fn, patched_object)
else:
patched = _get_sync_patched(fn, patched_object)
patched_object.patched = patched
_patched_registry.register_callable(patched_object)
return patched
```
Теперь взглянем на саму подставновку зависимостей провайдера Factory и FactoryAggregate
```python
cdef inline object __factory_call(Factory self, tuple args, dict kwargs):
cdef object instance
instance = __call(
self.__instantiator.__provides,
args,
self.__instantiator.__args,
self.__instantiator.__args_len,
kwargs,
self.__instantiator.__kwargs,
self.__instantiator.__kwargs_len,
self.__async_mode,
)
if self.__attributes_len > 0:
attributes = __provide_attributes(self.__attributes, self.__attributes_len)
is_future_instance = __is_future_or_coroutine(instance)
is_future_attributes = __is_future_or_coroutine(attributes)
if is_future_instance or is_future_attributes:
future_instance = instance if is_future_instance else __future_result(instance)
future_attributes = attributes if is_future_attributes else __future_result(attributes)
return __async_inject_attributes(future_instance, future_attributes)
__inject_attributes(instance, attributes)
return instance
```
```python
cdef class FactoryAggregate(Aggregate):
@property
def factories(self):
"""Return dictionary of factories, read-only.
Alias for ``.providers()`` attribute.
"""
return self.providers
def set_factories(self, factory_dict=None, **factory_kwargs):
"""Set factories.
Alias for ``.set_providers()`` method.
"""
return self.set_providers(factory_dict, **factory_kwargs)
```
В данном случае провайдер разделает асинхронную реализацию от синхронной и по разному отдаёт объект зависимости.
Что же происходит при использовании FactoryAggregate
1. В момент вызова зависимости реализуются все фабрики внутри неё?
2. Как подставляется обёртка FactoryAggregate внутрь нашего объекта?
3. Почему фабрика с асинхронным объектом не реализуется до конца, а остаётся корутиной?
Ещё раз читаем код:
```python
cdef inline object __async_inject_attributes(future_instance, future_attributes):
future_result = asyncio.Future()
attributes_ready = asyncio.gather(future_instance, future_attributes)
attributes_ready.add_done_callback(
functools.partial(
__async_inject_attributes_callback,
future_result,
),
)
asyncio.ensure_future(attributes_ready)
return future_result
cdef inline void __inject_attributes(object instance, dict attributes):
for name, value in attributes.items():
setattr(instance, name, value)
```
Понимаем, что имеем дело с синхронной зависимостью всегда (FactoryAggregate можно считать провайдером dict'а), поэтому при вызове нашего экземпляра FactoryAggregate она выдаёт отдаёт либо объект реализованной фабрики, либо Future, которую необходимо до реализовать самостоятельно. То есть если хотя-бы одна асинхронная зависимость есть в фабрике, то мы получим Future на выходе.
Как сделать, чтобы всегда было одинаковое поведение?
Давайте попробуем для всех фабрик, которые мы используем создать асинхронную обёртку. Если фабрика возвращает объект Cotutine или Future, то после await операции - возвратили результат самой фабрики. А если пришёл объект фабрики использующейся в шаблонах, то await для фабрики тоже отработал и вернул сам себя.
Напишем базовый класс:
```python
from abc import ABC
class LoaderFactoryAggregateMixin(ABC):
"""Штука позволяет безболезненно получить содержимое FactoryAggregate без
if asyncio.isfuture() or asyncio.iscoroutine()"""
def __await__(self):
async def wrapper():
return self
return wrapper().__await__()
```
Подключим к двум реализациям
```plantuml
@startuml
skinparam linetype ortho
abstract class "LoaderFactoryAggregateMixin" as LF
interface "UserRepositoryBase" as UB
class "UserExternalRepository" as C2
class "UserDatabaseRepository" as C3
class "UseCase" as H
C2 --|> UB
C2 --|> LF
C3 --|> UB
C3 --|> LF
H <-- UB
@enduml
```
Запускаем. Работает
*Примечание. Если вы уверены в том, что все зависимости фабрики получаются синхронно, не используют под собой provide.Resource с асинхронными функциями или provide.Coroutine - костыль не нужен*
Sign in with Wallet
Connect another wallet