В этой статье я хотел бы представить di , структура впрыска зависимостей, которая стремится быть:
- Интуитивно понятный : Простые вещи легко, сложные вещи возможны
- Лаконичный : Низкий шаблон, может запустить ваш код непосредственно через автоматическое
- Правильно : Протестировано с Mypy ; Экспортирует действительные аннотации типа
- Мощный : С жизненными веществами, общими зависимостями и гибкий API
- Исполнитель : Поддержка асинхронных зависимостей и одновременного исполнения
Введение в инъекцию зависимостей
Инъекция зависимости – это метод для организации более крупных приложений путем передачи контроля ваших зависимостей на «что -то» (часто структура), которая собирает их для вас. Это сложная тема, так что, возможно, лучше просто посмотреть на пример. Мы создадим наш пример вокруг довольно распространенного сценария: у вас есть класс, который должен выполнять веб -запросы, поэтому ему нужен клиент HTTP (мы будем использовать HTTPX).
from httpx import Client
class WeatherClient:
def __init__(self) -> None:
self.client = Client()
def get_temp(self, lat: float, long: float) -> float:
self.client.get(...)
...
Этот класс делает не Используйте инъекцию зависимости. Проблема возникает, когда мы хотим проверить этот класс: мы, вероятно, не хотим делать реальные веб -запросы в нашем модульном тесте.
К счастью, HTTPX предлагает возможность использовать Мик транспорт :
import httpx
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"temp": 43.21})
client = httpx.Client(transport=httpx.MockTransport(handler))
Но у нас есть проблема: как мы получаем наш Клиент Использование фиктивного транспорта в нашем WeatherClient ? Один из них часто используется Обезьяна исправления Анкет В то время как Monkey Patching имеет свои варианты использования, это также может привести к множеству очень запутанных проблем и грязного кода, особенно в более крупном проекте.
Инъекция зависимости дает нам альтернативу: WeatherClient принять экземпляр из httpx. Клиент :
from typing import Optional
from httpx import Client
class WeatherClient:
def __init__(self, client: Optional[Client] = None) -> None:
self.client = client or Client()
Эта форма инъекции зависимостей называется Инъекция конструктора Потому что мы внедряем зависимость в конструкторе/ __init__ метод
В этом случае мы решили добавить Нет по умолчанию, чтобы мы могли обнаружить, когда вызывающий абонент не включил Клиент И поэтому мы можем сделать по умолчанию один. Это во многом вопрос предпочтения, и зависит от проекта, над которым вы работаете. Для библиотеки это, вероятно, хорошая идея. Для приложения, возможно, не так много.
Примечание : Вы заметили, что httpx. Клиент Сам использует инъекцию зависимостей для Транспорт параметр ? Окрашенный, да!
Проблемы
Как и выше, инъекция зависимостей не так сумасшедшая, как кажется, это может быть просто. И вы, вероятно, уже используете его в библиотеках, таких как httpx Анкет
Но каковы недостатки?
Что ж, основная проблема заключается в том, что, когда вы получите глубоко вложенные/много зависимостей, как это, как правило, в более крупном проекте, вы можете получить некоторые действительно уродливые вызовы конструктора, или вам может понадобиться такая же зависимость в нескольких местах , что означает, что вам нужна временная переменная и т. Д.
По сути, код в собрать И отслеживать все эти зависимости может стать головной болью сама по себе.
Введите: рамки впрыска зависимостей.
Рамки впрыска в зависимости стремятся решить эту проблему, абстрагируя шаблон создания и соединения зависимостей вместе.
Организации впрыска зависимостей
Хотя рамки впрыска зависимостей бывают разных форм и размеров, в принципе все они обеспечивают аналогичную функциональность:
- Зарегистрируйте зависимости и поставщики этих зависимостей (то есть сообщите структуре, как построить свои зависимости)
- Введите свои зависимости для вас
Некоторые фреймворки также имеют более продвинутые функции, такие как:
- Автофихание: транзитивные зависимости обнаруживаются из имен параметров или аннотаций типа
- Жизненные циклы: способность подключаться к созданию, удалению и ошибкам во время жизненного цикла вашей зависимости.
- Области: синглтонские зависимости или зависимости, которые повторно используются в каждом вызове, если они необходимы в нескольких местах.
di является одной из последних рамок, предлагающих все эти функции и многое другое.
Используя ди
Начнем с простого примера. Мы хотим создать веб -приложение, которое использует наше WeatherClient Анкет
import os
from dataclasses import dataclass, field
from xyz import App
def url_from_env() -> str:
return os.environ["WEATHER_URL"]
@dataclass
class AppConfig:
weather_url: str = field(default_factory=url_from_env)
class MyApp:
def __init__(self, weather_client: WeatherClient) -> None:
self.weather_client = weather_client
Создание этого вручную может выглядеть примерно как:
app = App(
weather_client=WeatherClient(
client=httpx.Client(
...
)
),
config=AppConfig(
...
)
)
Используя di :
from di import Container, Dependant
container = Container()
app = container.execute_sync(
container.solve(Dependant(App))
)
Обратите внимание, что не имеет значения, насколько глубоки зависимости или сколько их, мы только когда -либо говорили: «Мы хотим экземпляр приложения».
Как насчет инъекции Клиент ?
client = Client(
transport=MockTransport(handler=handler)
)
container.bind(Dependant(lambda: client), Client)
app = container.execute_sync(
container.solve(Dependant(App))
)
Опять же, важна здесь, это Клиент Может быть вложенные 10 слоев зависимостей глубоко, и приведенный выше код вообще не изменится.
Расширенная функциональность
Помимо основного использования выше, di Поддерживать намного больше функциональности.
Вот некоторые фрагменты, демонстрирующие функциональность, но для полных примеров и дополнительных объяснений см. В нашем Документы Анкет
Распределение зависимости
from di import Container, Depends, Dependant
def func() -> object:
return object()
def dependant(
one: object = Depends(func),
two: object = Depends(func),
three: object = Depends(func, share=False)
) -> None:
assert one is two and two is not three
container = Container()
container.execute_sync(container.solve(Dependant(dependant)))
В этом примере Func называется дважды (по умолчанию в двух отдельных потоках), один раз для один и два , которые имеют одинаковое значение, и один раз для Три Анкет
Прицелы и жизненные циклы
from typing import Generator
from di import Container, Depends, Dependant
def func() -> Generator[int, None, None]:
print("func startup")
yield 1
print("func shutdown")
def dependant(v: int = Depends(func, scope=1234)) -> int:
print("computing")
return v + 1
container = Container()
with container.enter_local_scope(1234):
print("enter scope")
res = container.execute_sync(
container.solve(Dependant(dependant))
)
print("exit scope")
assert res == 2
Этот пример будет печатать:
enter scope func startup computing func shutdown exit scope
И эквивалентно:
from contextlib import contextmanager
with contextmanager(func)() as value:
dependant(v=value)
Прицелы в di являются произвольными: любая хешейная стоимость подойдет. Пока вы согласны в объявлении областей, di не будет заботиться о том, что вы используете. Если «вы» – это структура, вызывая код пользователя, это дает вам свободу определять и управлять своими собственными областями. Для веб -структуры это может быть «приложение» и «запрос». Для CLI может быть единственный «Call». Для TUI может быть «приложение» и «Объем событий»
Решенные зависимости
Часто вы должны запускать одну и ту же зависимость несколько раз, но не динамически не изменяют график зависимости каждый выполнение (вы бы динамически изменили график, если бы вы делали контейнер. Связь между каждым вызовом).
В этих сценариях di Позволяет сохранить весь вычисленный график зависимости в Сохраняющая защита и выполнить один и тот же график несколько раз. На самом деле, это то, что мы уже делали уже каждый раз, когда называли Контейнер.solve Анкет
# solving locks in binds, but not cached values or scopes solved = container.solve(Dependant(dependency)) res1 = container.execute_sync(solved) res2 = container.execute_sync(solved)
Асинхронное исполнение
Просто замените execute_sync для execute_async Для асинхронной поддержки:
from di import Container, Dependant
async def dep() -> int:
return 1
async def main() -> None:
container = Container()
solved = container.solve(Dependant(dep))
res = await container.execute_async(solved)
assert res == 1
Подобно синхронизации, асинхронизированное выполнение поддерживает произвольные вызовы, а также функции генератора, которые используются в качестве менеджеров контекста.
В зависимости от индивидуальной протокола
Disecidantprotocol Интерфейс обеспечивает абстрактное представление спецификации зависимости. Контейнер использует это абстрактное представление для построения графиков зависимости.
Это означает, что вы можете легко настроить много поведения в di переопределив этот API. Если вы хотите использовать часть встроенной функциональности и просто немного настроить ее, вы можете унаследовать от ди -дюймовый Зависимый В Но тебе не нужно.
Например, чтобы принять значение по умолчанию вместо Callable:
from di import Dependant
from di.types.providers import DependencyType
from di.types.scopes import Scope
class DefaultDependant(Dependant[DependencyType]):
def __init__(
self,
default: DependencyType,
scope: Scope,
shared: bool,
) -> None:
super().__init__(
call=lambda: default,
scope=scope,
share=share,
)
Теперь, если не будет привязки/переопределения зависимости, она будет выполняться так же, как если бы она была вызвана напрямую.
Исполнители
Один из принципов дизайна di это изолировать работу контейнера на различные шаги:
- Проводка, которую можно контролировать через зависимость
- Решение, которое принадлежит контейнеру
- Выполнение, которое управляется через API исполнителя
С этой целью di Позволяет вам предоставить своему собственному исполнителю, который должен взять список групп задач (вызов, которые не принимают аргументов) и выполнять их в указанном порядке.
API исполнителя выглядит следующим образом:
import typing
ResultType = typing.TypeVar("ResultType")
Task = typing.Union[
typing.Callable[[], None],
typing.Callable[[], typing.Awaitable[None]]
]
class SyncExecutor(typing.Protocol):
def execute_sync(
self,
tasks: typing.List[typing.List[Task]],
get_result: typing.Callable[[], ResultType],
) -> ResultType:
...
class AsyncExecutor(typing.Protocol):
async def execute_async(
self,
tasks: typing.List[typing.List[Task]],
get_result: typing.Callable[[], ResultType],
) -> ResultType:
...
Исполнитель по умолчанию ( di.executors. Defaultexecutor ) поддерживает как синхронизацию, так и асинхронное выполнение. Зависимости выполняются одновременно (через потоки в выполнении синхронизации и задачи в асинхронном выполнении) для зависимостей, которые могут быть выполнены одновременно.
Простая реализация Syncexecutor может выглядеть как:
class SimpleExecutor(SyncExecutor):
def execute_sync(
self,
tasks: typing.List[typing.List[Task]],
get_result: typing.Callable[[], ResultType],
) -> ResultType:
for group in tasks:
for task in group:
res = task()
if inspect.isawaitable(res):
raise TypeError
return get_result()
Обратите внимание, что Syncexecutor может быть вручено асинхронные функции в качестве задач. Синхронизированная/асинхронная часть относится к API самого исполнителя, а не к задачам, которую он может быть вручен. Реализация зависит от того, чтобы как -то поднять исключение или выполнить асинхронную задачу.
Интеграция в другие библиотеки
В то время как di Должно быть достаточно просто, чтобы его можно было использовать конечными пользователями, он сияет при интеграции в структуру или библиотеку, которая полностью охватывает инверсию управления.
С этой целью мы предоставляем Пример интеграции , в настоящее время только для Starlette и Textual, но приветствуются больше идей!
Производительность
Мы предоставляем несколько Основные тесты по сравнению с FASTAPI Анкет
Результаты сильно зависят от размера DAG.
Для относительно большого, но не совершенно абсурдного Дага из 12 вершин с каждой вершиной, имеющей 0-3 края, di Выработает немного быстрее, чем Fastapi (пара процентов за запрос).
Для больших даг, особенно где di способен выполнять зависимости одновременно, di может быть в 10 раз быстрее, чем FASTAPI. Чтобы быть справедливым, это не реальный сценарий, и Fastapi делает намного больше, чем Starlette di Интеграция делает (отслеживание вещей для OpenAPI и т. Д.). Если эти тесты несправедливы по другим причинам, пожалуйста, дайте мне знать.
Вывод
Инъекция зависимости и инверсия контроля являются мощными методами, но она может усложняться. Python снова и снова показывает, что он может занять сложные вещи и сделать их простыми для пользователей. Fastapi – один из лучших примеров этого.
Я надеюсь, что di может помочь другим библиотекам, структурам или проектам использовать инверсию контроля и инъекции зависимостей, не проходя через боль от написания собственной системы впрыска зависимостей.
Оригинал: “https://dev.to/adriangb/di-pythonic-dependency-injection-5b27”