Мониторинг демон на Asyncio + Dependency Injector — руководство по применению dependency injection

не согласен, DI очень интересно послучилось.
я года 3 назад что то похожее писал для своих задач, правда там создание экземпляра класса и параметра конструктора декларировались на уровне конфигурации, но это уже не суть важно.

абсолютно непонятно зачем вообще это все нужно и какую вообще проблему решает DI

Это классный вопрос. Специально его не затрагивал. Хотел сделать практическое руководство.

Про пользу dependency injection. Я относился к нему пренебрежительно пока в 2014 на себе не почувствовал его магическую силу. Это случилось при рефакторинге крупной легаси платформы. Было много запутанного кода. Мы применили DI. Когда применяешь DI все связи становятся явными. Если когда-то пробовали вышивать, это как заглянуть на обратную сторону рисунка. Структура приложения вырисовывается сама собой. Она отделяется от runtime части и с ней становится удобно работать. После применения DI все стало прозрачно и просто. В тот момент я осознал силу подхода.

Так и появился Dependency Injector.

Если интересно, напишу отдельную статью.

PS: Подход неинтуитивный, нужно привыкнуть. Плюс это инвестиция в будущее: когда пишешь приложение и добавляешь все в контейнер — кажется избыточно. Когда возвращаешься через время чтобы что-то поменять — рад тому, что контейнер есть.

что мешает сделать тоже самое в функции create_app и без всякого фреймворка? Результат будет одинаковый.

Мой фреймворк позволяет описать всю структуру приложения в декларативном стиле не создав ни одного объекта. В таком виде со структурой удобнее всего работать.

Тоже самое можно сделать и без него, только у вас получится такой же фреймворк. Его нужно будет протестировать, оптимизировать, настроить сборку 44 wheels для всех версий Python и ОС, поддерживать, фиксить баги, написать документацию, примеры и это руководство. Для моего фреймворка я это все уже сделал.

Писать все в функцию create_app() неправильно. Это смешивание декларативной и runtime частей. Такие приложения тяжелее понимать.

Понимаю. Постараюсь объяснить свою точку зрения.

Dependency Injector — это про 2 вещи:

1. Dependency injection — это хорошо.

Почему? Decoupling со всеми вытекающими обстоятельствами: гибкость, тестирование и т. д.

2. Указывать зависимости явно — тоже хорошо.

Приведу примеры от обратного.

Вариант №1. В Dependency Injector можно реализовать метод container.autowire(). Это метод сможет с высокой точностью связать все компоненты приложения без вас. Контейнер будет выглядеть следующим образов:

class ApplicationContainer(containers.DeclarativeContainer):

    config = providers.Configuration()

    configure_logging = providers.Callable(logging.basicConfig)

    http_client = providers.Factory(http.HttpClient)

    example_monitor = providers.Factory(monitors.HttpMonitor)

    httpbin_monitor = providers.Factory(monitors.HttpMonitor)

    dispatcher = providers.Factory(dispatcher.Dispatcher)

Много ли можно сказать о приложении по такому контейнеру? Почти ничего.

Вариант №2. Отказаться от наличия контейнера и добавить магию в стиле декоратора @inject:

from dependency_injector import inject

@inject
class Dispatcher:

    def __init__(self, monitors: List[Monitor]) -> None:
        ...

@inject
class HttpMonitor(Monitor):

    def __init__(
            self,
            http_client: HttpClient,
            options: Dict[str, Any],
    ) -> None:
        ...

В таком случае ваш код будет прочно привязан к dependency injection фреймворку. Как подключать сторонние модули..? Как собрать один и тот же класс с разными параметрами (пример с HttpMonitor из данного примера)..? Для простых случаев удобно, в более сложных — фреймворк наложит прилично ограничений.

В первой версии Dependency Injector декоратор @inject был. Убрал его специально. Считаю, что код должен жить отдельно от фреймворка. Это главный принцип Dependency Injector. Он не загрязняет ваш код магическими декораторами и правилами. Вместо этого он накладывается поверх на любой код, написанный по принципу dependency injection.

Про pytest

Фикстуры в pytest — удобная штука. Зависимости там тоже указываются явно. Если переписать контейнер из этого примера, то получится вот так:

@fixture
def config():
    return providers.Configuration()


@fixture
def configure_logging(config):
    return logging.basicConfig(
        stream=sys.stdout,
        level=config.log.level,
        format=config.log.format,
    )


@fixture
def http_client(config):
    return http.HttpClient()


@fixture
def example_monitor(http_client, config):
    return monitors.HttpMonitor(
        http_client=http_client,
        options=config.monitors.example,
    )


@fixture
def httpbin_monitor(http_client, config):
    return monitors.HttpMonitor(
        http_client=http_client,
        options=config.monitors.httpbin,
    )

@fixture
def dispatcher(example_monitor, httpbin_monitor):
    return dispatcher.Dispatcher(
        monitors=[
            example_monitor,
            httpbin_monitor,
        ],
    )

В целом тоже самое, но писать больше.

В открытом доступе есть вот такой пример:

"""Application containers module."""

from dependency_injector import containers, providers

from newsfeed import core, infrastructure, domainmodel, webapi


class Core(containers.DeclarativeContainer):
    """Core container."""

    config = providers.Configuration('core')

    configure_logging = providers.Callable(
        core.log.configure_logging,
        level=config.log_level,
    )

    configure_event_loop = providers.Callable(
        core.loop.configure_event_loop,
        enable_uvloop=config.enable_uvloop,
    )


class Infrastructure(containers.DeclarativeContainer):
    """Infrastructure container."""

    config = providers.Configuration('infrastructure')

    event_queue = providers.Singleton(
        infrastructure.event_queues.InMemoryEventQueue,
        config=config.event_queue,
    )

    event_storage = providers.Singleton(
        infrastructure.event_storages.RedisEventStorage,
        config=config.event_storage,
    )

    subscription_storage = providers.Singleton(
        infrastructure.subscription_storages.RedisSubscriptionStorage,
        config=config.subscription_storage,
    )


class DomainModel(containers.DeclarativeContainer):
    """Domain model container."""

    config = providers.Configuration('domainmodel')

    infra: Infrastructure = providers.DependenciesContainer()

    # Common

    newsfeed_id_specification = providers.Singleton(
        domainmodel.newsfeed_id.NewsfeedIDSpecification,
        max_length=config.newsfeed_id_length,
    )

    # Subscription

    subscription_factory = providers.Factory(
        domainmodel.subscription.SubscriptionFactory,
        cls=domainmodel.subscription.Subscription,
    )

    subscription_specification = providers.Singleton(
        domainmodel.subscription.SubscriptionSpecification,
        newsfeed_id_specification=newsfeed_id_specification,
    )

    subscription_repository = providers.Singleton(
        domainmodel.subscription.SubscriptionRepository,
        factory=subscription_factory,
        storage=infra.subscription_storage,
    )

    subscription_service = providers.Singleton(
        domainmodel.subscription.SubscriptionService,
        factory=subscription_factory,
        specification=subscription_specification,
        repository=subscription_repository,
    )

    # Event

    event_factory = providers.Factory(
        domainmodel.event.EventFactory,
        cls=domainmodel.event.Event,
    )

    event_specification = providers.Singleton(
        domainmodel.event.EventSpecification,
        newsfeed_id_specification=newsfeed_id_specification,
    )

    event_repository = providers.Singleton(
        domainmodel.event.EventRepository,
        factory=event_factory,
        storage=infra.event_storage,
    )

    event_dispatcher_service = providers.Singleton(
        domainmodel.event_dispatcher.EventDispatcherService,
        event_factory=event_factory,
        event_specification=event_specification,
        event_queue=infra.event_queue,
    )

    event_processor_service = providers.Singleton(
        domainmodel.event_processor.EventProcessorService,
        event_queue=infra.event_queue,
        event_factory=event_factory,
        event_repository=event_repository,
        subscription_repository=subscription_repository,
    )


class WebApi(containers.DeclarativeContainer):
    """Web API container."""

    config = providers.Configuration('webapi')

    domain: DomainModel = providers.DependenciesContainer()

    web_app = providers.Factory(
        webapi.app.create_web_app,
        base_path=config.base_path,
        routes=[
            # Subscriptions
            webapi.app.route(
                method='GET',
                path='/newsfeed/{newsfeed_id}/subscriptions/',
                handler=providers.Coroutine(
                    webapi.handlers.subscriptions.get_subscriptions_handler,
                    subscription_service=domain.subscription_service,
                ),
            ),
            webapi.app.route(
                method='POST',
                path='/newsfeed/{newsfeed_id}/subscriptions/',
                handler=providers.Coroutine(
                    webapi.handlers.subscriptions.post_subscription_handler,
                    subscription_service=domain.subscription_service,
                ),
            ),
            webapi.app.route(
                method='DELETE',
                path='/newsfeed/{newsfeed_id}/subscriptions/{subscription_id}/',
                handler=providers.Coroutine(
                    webapi.handlers.subscriptions.delete_subscription_handler,
                    subscription_service=domain.subscription_service,
                ),
            ),
            webapi.app.route(
                method='GET',
                path='/newsfeed/{newsfeed_id}/subscribers/subscriptions/',
                handler=providers.Coroutine(
                    webapi.handlers.subscriptions.get_subscriber_subscriptions_handler,
                    subscription_service=domain.subscription_service,
                ),
            ),

            # Events
            webapi.app.route(
                method='GET',
                path='/newsfeed/{newsfeed_id}/events/',
                handler=providers.Coroutine(
                    webapi.handlers.events.get_events_handler,
                    event_repository=domain.event_repository,
                ),
            ),
            webapi.app.route(
                method='POST',
                path='/newsfeed/{newsfeed_id}/events/',
                handler=providers.Coroutine(
                    webapi.handlers.events.post_event_handler,
                    event_dispatcher_service=domain.event_dispatcher_service,
                ),
            ),
            webapi.app.route(
                method='DELETE',
                path='/newsfeed/{newsfeed_id}/events/{event_id}/',
                handler=providers.Coroutine(
                    webapi.handlers.events.delete_event_handler,
                    event_dispatcher_service=domain.event_dispatcher_service,
                ),
            ),

            # Miscellaneous
            webapi.app.route(
                method='GET',
                path='/status/',
                handler=providers.Coroutine(
                    webapi.handlers.misc.get_status_handler,
                ),
            ),
            webapi.app.route(
                method='GET',
                path='/docs/',
                handler=providers.Coroutine(
                    webapi.handlers.misc.get_openapi_schema_handler,
                    base_path=config.base_path,
                ),
            ),
        ],
    )

Тут используется несколько контейнеров.

Сборка выглядит вот так:

class Application:
    """Application."""

    class Containers:
        """Application containers."""

        core = Core
        infrastructure = Infrastructure
        domainmodel = DomainModel
        webapi = WebApi

    def __init__(self, config: Dict[str, Any]):
        """Initialize application."""
        self.config = config

        self.core: Core = self.Containers.core()
        self.core.config.override(self.config.get(self.core.config.get_name()))

        self.infrastructure: Infrastructure = self.Containers.infrastructure()
        self.infrastructure.config.override(self.config.get(self.infrastructure.config.get_name()))

        self.domainmodel: DomainModel = self.Containers.domainmodel()
        self.domainmodel.config.override(self.config.get(self.domainmodel.config.get_name()))
        self.domainmodel.infra.override(self.infrastructure)

        self.webapi: WebApi = self.Containers.webapi()
        self.webapi.config.override(self.config.get(self.webapi.config.get_name()))
        self.webapi.domain.override(self.domainmodel)

        self.processor_tasks: List[asyncio.Task[None]] = []

    def main(self) -> None:
        """Run application."""
        self.core.configure_logging()
        self.core.configure_event_loop()

        web_app: web.Application = self.webapi.web_app()

        web_app.on_startup.append(self._start_background_tasks)
        web_app.on_cleanup.append(self._cleanup_background_tasks)

        web.run_app(web_app, port=int(self.webapi.config.port()), print=None)

Это немного устаревший вариант. Для такой сборки сейчас есть специальный провайдер Container. Он автоматизирует то, что написано в __init__().

Проект находится тут — Newsfeed.