Monitoramento daemon no Asyncio + Dependency Injector - tutorial de injeção de dependĂȘncia

OlĂĄ,



sou o criador do injetor de dependĂȘncia . Esta Ă© uma estrutura de injeção de dependĂȘncia para Python.



Este Ă© outro tutorial para construir aplicativos com o Dependency Injector.



Hoje eu quero mostrar como vocĂȘ pode construir um daemon assĂ­ncrono baseado em um mĂłdulo asyncio.



O manual consiste nas seguintes partes:



  1. O que vamos construir?
  2. Verificação de ferramenta
  3. Estrutura do projeto
  4. Preparando o ambiente
  5. Registro e configuração
  6. Expedidor
  7. Monitorando example.com
  8. Monitorando httpbin.org
  9. Testes
  10. ConclusĂŁo


O projeto concluĂ­do pode ser encontrado no Github .



Para começar, é desejåvel ter:



  • Conhecimento inicial de asyncio
  • Compreender o princĂ­pio da injeção de dependĂȘncia


O que vamos construir?



Estaremos construindo um daemon de monitoramento que irå monitorar o acesso aos serviços da web.



O daemon enviarå solicitaçÔes para example.com e httpbin.org a cada poucos segundos. Ao receber uma resposta, ele gravarå os seguintes dados no log:



  • CĂłdigo de resposta
  • NĂșmero de bytes em resposta
  • Tempo necessĂĄrio para concluir a solicitação






Verificação de ferramenta



Estaremos usando Docker e docker-compose . Vamos verificar se eles estĂŁo instalados:



docker --version
docker-compose --version


A saĂ­da deve ser semelhante a esta:



Docker version 19.03.12, build 48a66213fe
docker-compose version 1.26.2, build eefe0d31


Se Docker ou docker-compose nĂŁo estiverem instalados, eles precisam ser instalados antes de continuar. Siga estes guias:





As ferramentas estĂŁo prontas. Vamos passar para a estrutura do projeto.



Estrutura do projeto



Crie uma pasta de projeto e vå até ela:



mkdir monitoring-daemon-tutorial
cd monitoring-daemon-tutorial


Agora precisamos criar uma estrutura inicial do projeto. Crie arquivos e pastas seguindo a estrutura abaixo. Todos os arquivos estarĂŁo vazios por enquanto. NĂłs os preencheremos mais tarde.



Estrutura inicial do projeto:



./
├── monitoringdaemon/
│   ├── __init__.py
│   ├── __main__.py
│   └── containers.py
├── config.yml
├── docker-compose.yml
├── Dockerfile
└── requirements.txt


A estrutura inicial do projeto estå pronta. Expandiremos sobre isso nas seçÔes seguintes.



A seguir, estamos aguardando a preparação do ambiente.



Preparando o ambiente



Nesta seção, prepararemos o ambiente para iniciar nosso daemon.



Primeiro vocĂȘ precisa definir dependĂȘncias. Usaremos pacotes como este:



  • dependency-injector - framework de injeção de dependĂȘncia
  • aiohttp - estrutura da web (sĂł precisamos de um cliente http)
  • pyyaml - biblioteca para analisar arquivos YAML, usada para ler a configuração
  • pytest - estrutura de teste
  • pytest-asyncio- biblioteca auxiliar para asyncioaplicativos de teste
  • pytest-cov - biblioteca auxiliar para medir a cobertura de cĂłdigo por testes


Vamos adicionar as seguintes linhas ao arquivo requirements.txt:



dependency-injector
aiohttp
pyyaml
pytest
pytest-asyncio
pytest-cov


E execute no terminal:



pip install -r requirements.txt


Em seguida, criamos Dockerfile. Ele descreverå o processo de construção e inicialização de nosso daemon. Vamos uså-lo python:3.8-bustercomo imagem de base.



Vamos adicionar as seguintes linhas ao arquivo Dockerfile:



FROM python:3.8-buster

ENV PYTHONUNBUFFERED=1

WORKDIR /code
COPY . /code/

RUN apt-get install openssl \
 && pip install --upgrade pip \
 && pip install -r requirements.txt \
 && rm -rf ~/.cache

CMD ["python", "-m", "monitoringdaemon"]


A Ășltima etapa Ă© definir as configuraçÔes docker-compose.



Vamos adicionar as seguintes linhas ao arquivo docker-compose.yml:



version: "3.7"

services:

  monitor:
    build: ./
    image: monitoring-daemon
    volumes:
      - "./:/code"


Tudo estå pronto. Vamos começar a construir a imagem e verificar se o ambiente estå configurado corretamente.



Vamos executar no terminal:



docker-compose build


O processo de construção pode levar vĂĄrios minutos. No final, vocĂȘ verĂĄ:



Successfully built 5b4ee5e76e35
Successfully tagged monitoring-daemon:latest


ApĂłs a conclusĂŁo do processo de criação, inicie o contĂȘiner:



docker-compose up


VocĂȘ verĂĄ:



Creating network "monitoring-daemon-tutorial_default" with the default driver
Creating monitoring-daemon-tutorial_monitor_1 ... done
Attaching to monitoring-daemon-tutorial_monitor_1
monitoring-daemon-tutorial_monitor_1 exited with code 0


O ambiente estĂĄ pronto. O contĂȘiner inicia e sai com o cĂłdigo 0.



A próxima etapa é configurar o registro e a leitura do arquivo de configuração.



Registro e configuração



Nesta seção, configuraremos o registro e a leitura do arquivo de configuração.



Vamos começar adicionando a parte principal de nosso aplicativo - o contĂȘiner de dependĂȘncia (mais adiante, apenas o contĂȘiner). O contĂȘiner conterĂĄ todos os componentes do aplicativo.



Vamos adicionar os dois primeiros componentes. Este é um objeto de configuração e uma função para configurar o registro.



Vamos editar containers.py:



"""Application containers module."""

import logging
import sys

from dependency_injector import containers, providers


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    config = providers.Configuration()

    configure_logging = providers.Callable(
        logging.basicConfig,
        stream=sys.stdout,
        level=config.log.level,
        format=config.log.format,
    )


Usamos os parùmetros de configuração antes de definir seus valores. Este é o princípio pelo qual o provedor trabalha Configuration.



Primeiro usamos, depois definimos os valores.



As configuraçÔes de registro estarão contidas no arquivo de configuração.



Vamos editar config.yml:



log:
  level: "INFO"
  format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s"


Agora vamos definir uma função que iniciarĂĄ nosso daemon. Ela geralmente Ă© chamada main(). Isso criarĂĄ um contĂȘiner. O contĂȘiner serĂĄ usado para ler o arquivo de configuração e chamar a função de configuraçÔes de registro.



Vamos editar __main__.py:



"""Main module."""

from .containers import ApplicationContainer


def main() -> None:
    """Run the application."""
    container = ApplicationContainer()

    container.config.from_yaml('config.yml')
    container.configure_logging()


if __name__ == '__main__':
    main()


O contĂȘiner Ă© o primeiro objeto do aplicativo. É usado para obter todos os outros objetos.


O registro e a leitura da configuração estão configurados. Na próxima seção, criaremos um gerenciador de tarefas de monitoramento.



Expedidor



É hora de adicionar um gerenciador de tarefas de monitoramento.



O despachante conterĂĄ uma lista de tarefas de monitoramento e controlarĂĄ sua execução. Ele executarĂĄ cada tarefa de acordo com a programação. Classe Monitor- classe base para tarefas de monitoramento. Para criar tarefas especĂ­ficas, vocĂȘ precisa adicionar classes filhas e implementar o mĂ©todo check().





Vamos adicionar um despachante e uma classe base para a tarefa de monitoramento.



Vamos criar dispatcher.pye monitors.pyno pacote monitoringdaemon:



./
├── monitoringdaemon/
│   ├── __init__.py
│   ├── __main__.py
│   ├── containers.py
│   ├── dispatcher.py
│   └── monitors.py
├── config.yml
├── docker-compose.yml
├── Dockerfile
└── requirements.txt


Vamos adicionar as seguintes linhas ao arquivo monitors.py:



"""Monitors module."""

import logging


class Monitor:

    def __init__(self, check_every: int) -> None:
        self.check_every = check_every
        self.logger = logging.getLogger(self.__class__.__name__)

    async def check(self) -> None:
        raise NotImplementedError()


e para o arquivo dispatcher.py:



""""Dispatcher module."""

import asyncio
import logging
import signal
import time
from typing import List

from .monitors import Monitor


class Dispatcher:

    def __init__(self, monitors: List[Monitor]) -> None:
        self._monitors = monitors
        self._monitor_tasks: List[asyncio.Task] = []
        self._logger = logging.getLogger(self.__class__.__name__)
        self._stopping = False

    def run(self) -> None:
        asyncio.run(self.start())

    async def start(self) -> None:
        self._logger.info('Starting up')

        for monitor in self._monitors:
            self._monitor_tasks.append(
                asyncio.create_task(self._run_monitor(monitor)),
            )

        asyncio.get_event_loop().add_signal_handler(signal.SIGTERM, self.stop)
        asyncio.get_event_loop().add_signal_handler(signal.SIGINT, self.stop)

        await asyncio.gather(*self._monitor_tasks, return_exceptions=True)

        self.stop()

    def stop(self) -> None:
        if self._stopping:
            return

        self._stopping = True

        self._logger.info('Shutting down')
        for task, monitor in zip(self._monitor_tasks, self._monitors):
            task.cancel()
        self._logger.info('Shutdown finished successfully')

    @staticmethod
    async def _run_monitor(monitor: Monitor) -> None:
        def _until_next(last: float) -> float:
            time_took = time.time() - last
            return monitor.check_every - time_took

        while True:
            time_start = time.time()

            try:
                await monitor.check()
            except asyncio.CancelledError:
                break
            except Exception:
                monitor.logger.exception('Error executing monitor check')

            await asyncio.sleep(_until_next(last=time_start))


O despachante precisa ser adicionado ao contĂȘiner.



Vamos editar containers.py:



"""Application containers module."""

import logging
import sys

from dependency_injector import containers, providers

from . import dispatcher


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    config = providers.Configuration()

    configure_logging = providers.Callable(
        logging.basicConfig,
        stream=sys.stdout,
        level=config.log.level,
        format=config.log.format,
    )

    dispatcher = providers.Factory(
        dispatcher.Dispatcher,
        monitors=providers.List(
            # TODO: add monitors
        ),
    )


Cada componente Ă© adicionado ao contĂȘiner.


Finalmente, precisamos atualizar a função main(). Pegaremos o despachante do contĂȘiner e chamaremos seu mĂ©todo run().



Vamos editar __main__.py:



"""Main module."""

from .containers import ApplicationContainer


def main() -> None:
    """Run the application."""
    container = ApplicationContainer()

    container.config.from_yaml('config.yml')
    container.configure_logging()

    dispatcher = container.dispatcher()
    dispatcher.run()


if __name__ == '__main__':
    main()


Agora vamos iniciar o daemon e verificar seu funcionamento.



Vamos executar no terminal:



docker-compose up


A saĂ­da deve ser semelhante a esta:



Starting monitoring-daemon-tutorial_monitor_1 ... done
Attaching to monitoring-daemon-tutorial_monitor_1
monitor_1  | [2020-08-08 16:12:35,772] [INFO] [Dispatcher]: Starting up
monitor_1  | [2020-08-08 16:12:35,774] [INFO] [Dispatcher]: Shutting down
monitor_1  | [2020-08-08 16:12:35,774] [INFO] [Dispatcher]: Shutdown finished successfully
monitoring-daemon-tutorial_monitor_1 exited with code 0


Tudo funciona corretamente. O despachante inicia e para, pois nĂŁo hĂĄ tarefas de monitoramento.



No final desta seção, o esqueleto de nosso demÎnio estå pronto. Na próxima seção, adicionaremos a primeira tarefa de monitoramento.



Monitorando example.com



Nesta seção, adicionaremos uma tarefa de monitoramento que monitorarå o acesso a http://example.com .



Começaremos estendendo nosso modelo de classe com um novo tipo de tarefa de monitoramento HttpMonitor.



HttpMonitoré uma aula infantil Monitor. Implementaremos o método check (). Ele enviarå uma solicitação HTTP e registrarå a resposta recebida. Os detalhes da solicitação HTTP serão delegados à classe HttpClient.





Vamos adicionar primeiro HttpClient.



Vamos criar um arquivo http.pyem um pacote monitoringdaemon:



./
├── monitoringdaemon/
│   ├── __init__.py
│   ├── __main__.py
│   ├── containers.py
│   ├── dispatcher.py
│   ├── http.py
│   └── monitors.py
├── config.yml
├── docker-compose.yml
├── Dockerfile
└── requirements.txt


E adicione as seguintes linhas a ele:



"""Http client module."""

from aiohttp import ClientSession, ClientTimeout, ClientResponse


class HttpClient:

    async def request(self, method: str, url: str, timeout: int) -> ClientResponse:
        async with ClientSession(timeout=ClientTimeout(timeout)) as session:
            async with session.request(method, url) as response:
                return response


Em seguida, vocĂȘ precisa adicionar HttpClientao contĂȘiner.



Vamos editar containers.py:



"""Application containers module."""

import logging
import sys

from dependency_injector import containers, providers

from . import http, dispatcher


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    config = providers.Configuration()

    configure_logging = providers.Callable(
        logging.basicConfig,
        stream=sys.stdout,
        level=config.log.level,
        format=config.log.format,
    )

    http_client = providers.Factory(http.HttpClient)

    dispatcher = providers.Factory(
        dispatcher.Dispatcher,
        monitors=providers.List(
            # TODO: add monitors
        ),
    )


Agora estamos prontos para adicionar HttpMonitor. Vamos adicionĂĄ-lo ao mĂłdulo monitors.



Vamos editar monitors.py:



"""Monitors module."""

import logging
import time
from typing import Dict, Any

from .http import HttpClient


class Monitor:

    def __init__(self, check_every: int) -> None:
        self.check_every = check_every
        self.logger = logging.getLogger(self.__class__.__name__)

    async def check(self) -> None:
        raise NotImplementedError()


class HttpMonitor(Monitor):

    def __init__(
            self,
            http_client: HttpClient,
            options: Dict[str, Any],
    ) -> None:
        self._client = http_client
        self._method = options.pop('method')
        self._url = options.pop('url')
        self._timeout = options.pop('timeout')
        super().__init__(check_every=options.pop('check_every'))

    @property
    def full_name(self) -> str:
        return '{0}.{1}(url="{2}")'.format(__name__, self.__class__.__name__, self._url)

    async def check(self) -> None:
        time_start = time.time()

        response = await self._client.request(
            method=self._method,
            url=self._url,
            timeout=self._timeout,
        )

        time_end = time.time()
        time_took = time_end - time_start

        self.logger.info(
            'Response code: %s, content length: %s, request took: %s seconds',
            response.status,
            response.content_length,
            round(time_took, 3)
        )


Estamos prontos para adicionar o cheque para http://example.com . Precisamos fazer duas alteraçÔes no contĂȘiner:



  • Adicione uma fĂĄbrica example_monitor.
  • Transfira example_monitorpara o despachante.


Vamos editar containers.py:



"""Application containers module."""

import logging
import sys

from dependency_injector import containers, providers

from . import http, monitors, dispatcher


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    config = providers.Configuration()

    configure_logging = providers.Callable(
        logging.basicConfig,
        stream=sys.stdout,
        level=config.log.level,
        format=config.log.format,
    )

    http_client = providers.Factory(http.HttpClient)

    example_monitor = providers.Factory(
        monitors.HttpMonitor,
        http_client=http_client,
        options=config.monitors.example,
    )

    dispatcher = providers.Factory(
        dispatcher.Dispatcher,
        monitors=providers.List(
            example_monitor,
        ),
    )


O provedor example_monitordepende dos valores de configuração. Vamos adicionar estes valores:



Editar config.yml:



log:
  level: "INFO"
  format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s"

monitors:

  example:
    method: "GET"
    url: "http://example.com"
    timeout: 5
    check_every: 5


Tudo estĂĄ pronto. Iniciamos o daemon e verificamos o trabalho.



Executamos no terminal:



docker-compose up


E vemos uma conclusĂŁo semelhante:



Starting monitoring-daemon-tutorial_monitor_1 ... done
Attaching to monitoring-daemon-tutorial_monitor_1
monitor_1  | [2020-08-08 17:06:41,965] [INFO] [Dispatcher]: Starting up
monitor_1  | [2020-08-08 17:06:42,033] [INFO] [HttpMonitor]: Check
monitor_1  |     GET http://example.com
monitor_1  |     response code: 200
monitor_1  |     content length: 648
monitor_1  |     request took: 0.067 seconds
monitor_1  |
monitor_1  | [2020-08-08 17:06:47,040] [INFO] [HttpMonitor]: Check
monitor_1  |     GET http://example.com
monitor_1  |     response code: 200
monitor_1  |     content length: 648
monitor_1  |     request took: 0.073 seconds


Nosso daemon pode monitorar a disponibilidade de acesso a http://example.com .



Vamos adicionar monitoramento https://httpbin.org .



Monitorando httpbin.org



Nesta seção, adicionaremos uma tarefa de monitoramento que monitorarå o acesso a http://example.com .



Adicionar uma tarefa de monitoramento para https://httpbin.org serĂĄ mais fĂĄcil porque todos os componentes estĂŁo prontos. Precisamos apenas adicionar um novo provedor ao contĂȘiner e atualizar a configuração.



Vamos editar containers.py:



"""Application containers module."""

import logging
import sys

from dependency_injector import containers, providers

from . import http, monitors, dispatcher


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    config = providers.Configuration()

    configure_logging = providers.Callable(
        logging.basicConfig,
        stream=sys.stdout,
        level=config.log.level,
        format=config.log.format,
    )

    http_client = providers.Factory(http.HttpClient)

    example_monitor = providers.Factory(
        monitors.HttpMonitor,
        http_client=http_client,
        options=config.monitors.example,
    )

    httpbin_monitor = providers.Factory(
        monitors.HttpMonitor,
        http_client=http_client,
        options=config.monitors.httpbin,
    )

    dispatcher = providers.Factory(
        dispatcher.Dispatcher,
        monitors=providers.List(
            example_monitor,
            httpbin_monitor,
        ),
    )


Vamos editar config.yml:



log:
  level: "INFO"
  format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s"

monitors:

  example:
    method: "GET"
    url: "http://example.com"
    timeout: 5
    check_every: 5

  httpbin:
    method: "GET"
    url: "https://httpbin.org/get"
    timeout: 5
    check_every: 5


Vamos iniciar o daemon e verificar os logs.



Vamos executar no terminal:



docker-compose up


E vemos uma conclusĂŁo semelhante:



Starting monitoring-daemon-tutorial_monitor_1 ... done
Attaching to monitoring-daemon-tutorial_monitor_1
monitor_1  | [2020-08-08 18:09:08,540] [INFO] [Dispatcher]: Starting up
monitor_1  | [2020-08-08 18:09:08,618] [INFO] [HttpMonitor]: Check
monitor_1  |     GET http://example.com
monitor_1  |     response code: 200
monitor_1  |     content length: 648
monitor_1  |     request took: 0.077 seconds
monitor_1  |
monitor_1  | [2020-08-08 18:09:08,722] [INFO] [HttpMonitor]: Check
monitor_1  |     GET https://httpbin.org/get
monitor_1  |     response code: 200
monitor_1  |     content length: 310
monitor_1  |     request took: 0.18 seconds
monitor_1  |
monitor_1  | [2020-08-08 18:09:13,619] [INFO] [HttpMonitor]: Check
monitor_1  |     GET http://example.com
monitor_1  |     response code: 200
monitor_1  |     content length: 648
monitor_1  |     request took: 0.066 seconds
monitor_1  |
monitor_1  | [2020-08-08 18:09:13,681] [INFO] [HttpMonitor]: Check
monitor_1  |     GET https://httpbin.org/get
monitor_1  |     response code: 200
monitor_1  |     content length: 310
monitor_1  |     request took: 0.126 seconds


A parte funcional estĂĄ concluĂ­da. O daemon monitora a disponibilidade de acesso a http://example.com e https://httpbin.org .



Na próxima seção, adicionaremos alguns testes.



Testes



Seria bom adicionar alguns testes. Vamos fazer isso.



Crie um arquivo tests.pyem um pacote monitoringdaemon:



./
├── monitoringdaemon/
│   ├── __init__.py
│   ├── __main__.py
│   ├── containers.py
│   ├── dispatcher.py
│   ├── http.py
│   ├── monitors.py
│   └── tests.py
├── config.yml
├── docker-compose.yml
├── Dockerfile
└── requirements.txt


e adicione as seguintes linhas a ele:



"""Tests module."""

import asyncio
import dataclasses
from unittest import mock

import pytest

from .containers import ApplicationContainer


@dataclasses.dataclass
class RequestStub:
    status: int
    content_length: int


@pytest.fixture
def container():
    container = ApplicationContainer()
    container.config.from_dict({
        'log': {
            'level': 'INFO',
            'formant': '[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s',
        },
        'monitors': {
            'example': {
                'method': 'GET',
                'url': 'http://fake-example.com',
                'timeout': 1,
                'check_every': 1,
            },
            'httpbin': {
                'method': 'GET',
                'url': 'https://fake-httpbin.org/get',
                'timeout': 1,
                'check_every': 1,
            },
        },
    })
    return container


@pytest.mark.asyncio
async def test_example_monitor(container, caplog):
    caplog.set_level('INFO')

    http_client_mock = mock.AsyncMock()
    http_client_mock.request.return_value = RequestStub(
        status=200,
        content_length=635,
    )

    with container.http_client.override(http_client_mock):
        example_monitor = container.example_monitor()
        await example_monitor.check()

    assert 'http://fake-example.com' in caplog.text
    assert 'response code: 200' in caplog.text
    assert 'content length: 635' in caplog.text


@pytest.mark.asyncio
async def test_dispatcher(container, caplog, event_loop):
    caplog.set_level('INFO')

    example_monitor_mock = mock.AsyncMock()
    httpbin_monitor_mock = mock.AsyncMock()

    with container.example_monitor.override(example_monitor_mock), \
            container.httpbin_monitor.override(httpbin_monitor_mock):

        dispatcher = container.dispatcher()
        event_loop.create_task(dispatcher.start())
        await asyncio.sleep(0.1)
        dispatcher.stop()

    assert example_monitor_mock.check.called
    assert httpbin_monitor_mock.check.called


Para executar os testes, execute no terminal:



docker-compose run --rm monitor py.test monitoringdaemon/tests.py --cov=monitoringdaemon


VocĂȘ deve obter um resultado semelhante:



platform linux -- Python 3.8.3, pytest-6.0.1, py-1.9.0, pluggy-0.13.1
rootdir: /code
plugins: asyncio-0.14.0, cov-2.10.0
collected 2 items

monitoringdaemon/tests.py ..                                    [100%]

----------- coverage: platform linux, python 3.8.3-final-0 -----------
Name                             Stmts   Miss  Cover
----------------------------------------------------
monitoringdaemon/__init__.py         0      0   100%
monitoringdaemon/__main__.py         9      9     0%
monitoringdaemon/containers.py      11      0   100%
monitoringdaemon/dispatcher.py      43      5    88%
monitoringdaemon/http.py             6      3    50%
monitoringdaemon/monitors.py        23      1    96%
monitoringdaemon/tests.py           37      0   100%
----------------------------------------------------
TOTAL                              129     18    86%


Observe como no teste test_example_monitorsubstituĂ­mos o HttpClientmock usando o mĂ©todo .override(). Dessa forma, vocĂȘ pode substituir o valor de retorno de qualquer provedor.



As mesmas açÔes são executadas no teste test_dispatcherpara substituir tarefas de monitoramento por simulaçÔes.





ConclusĂŁo



ConstruĂ­mos um daemon de monitoramento baseado no asyncioprincĂ­pio de injeção de dependĂȘncia. Usamos o Dependency Injector como uma estrutura de injeção de dependĂȘncia.



A vantagem que vocĂȘ obtĂ©m com o Dependency Injector Ă© o contĂȘiner.



O contĂȘiner começa a pagar quando vocĂȘ precisa entender ou alterar a estrutura do seu aplicativo. Com um contĂȘiner, Ă© fĂĄcil porque todos os componentes do aplicativo e suas dependĂȘncias estĂŁo em um sĂł lugar:



"""Application containers module."""

import logging
import sys

from dependency_injector import containers, providers

from . import http, monitors, dispatcher


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    config = providers.Configuration()

    configure_logging = providers.Callable(
        logging.basicConfig,
        stream=sys.stdout,
        level=config.log.level,
        format=config.log.format,
    )

    http_client = providers.Factory(http.HttpClient)

    example_monitor = providers.Factory(
        monitors.HttpMonitor,
        http_client=http_client,
        options=config.monitors.example,
    )

    httpbin_monitor = providers.Factory(
        monitors.HttpMonitor,
        http_client=http_client,
        options=config.monitors.httpbin,
    )

    dispatcher = providers.Factory(
        dispatcher.Dispatcher,
        monitors=providers.List(
            example_monitor,
            httpbin_monitor,
        ),
    )




Um contĂȘiner como um mapa de seu aplicativo. VocĂȘ sempre sabe o que depende do quĂȘ.



Qual Ă© o prĂłximo?






All Articles