Démon de surveillance sur Asyncio + Dependency Injector - Guide d'injection de dépendances

Salut,



je suis le créateur de Dependency Injector . Il s'agit d'un framework d'injection de dépendances pour Python.



Ceci est un autre didacticiel pour la création d'applications avec l'injecteur de dépendances.



Aujourd'hui, je veux montrer comment vous pouvez construire un démon asynchrone basé sur un module asyncio.



Le manuel comprend les parties suivantes:



  1. Qu'allons-nous construire?
  2. VĂ©rification de l'outil
  3. Structure du projet
  4. Préparer l'environnement
  5. Journalisation et configuration
  6. RĂ©partiteur
  7. Surveillance example.com
  8. Surveillance httpbin.org
  9. Des tests
  10. Conclusion


Le projet terminĂ© peut ĂȘtre trouvĂ© sur Github .



Pour commencer, il est souhaitable d'avoir:



  • Connaissance initiale de asyncio
  • Comprendre le principe de l'injection de dĂ©pendances


Qu'allons-nous construire?



Nous allons créer un démon de surveillance qui surveillera l'accÚs aux services Web.



Le dĂ©mon enverra des requĂȘtes Ă  example.com et httpbin.org toutes les quelques secondes. Lors de la rĂ©ception d'une rĂ©ponse, il Ă©crira les donnĂ©es suivantes dans le journal:



  • Code de rĂ©ponse
  • Nombre d'octets en rĂ©ponse
  • Temps nĂ©cessaire pour traiter la demande






VĂ©rification de l'outil



Nous utiliserons Docker et docker-compose . Vérifions qu'ils sont installés:



docker --version
docker-compose --version


La sortie devrait ressembler Ă  ceci:



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


Si Docker ou docker-compose ne sont pas installĂ©s, ils doivent ĂȘtre installĂ©s avant de continuer. Suivez ces guides:





Les outils sont prĂȘts. Passons Ă  la structure du projet.



Structure du projet



Créez un dossier de projet et accédez-y:



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


Nous devons maintenant créer une structure de projet initiale. Créez des fichiers et des dossiers en suivant la structure ci-dessous. Tous les fichiers seront vides pour le moment. Nous les remplirons plus tard.



Structure initiale du projet:



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


La structure initiale du projet est prĂȘte. Nous l'Ă©largirons dans les sections suivantes.



Ensuite, nous attendons la préparation de l'environnement.



Préparer l'environnement



Dans cette section, nous préparerons l'environnement pour démarrer notre démon.



Vous devez d'abord définir les dépendances. Nous utiliserons des packages comme celui-ci:



  • dependency-injector - framework d'injection de dĂ©pendances
  • aiohttp - framework web (nous n'avons besoin que d'un client http)
  • pyyaml - bibliothĂšque d'analyse des fichiers YAML, utilisĂ©e pour lire la configuration
  • pytest - cadre de test
  • pytest-asyncio- bibliothĂšque d'aide pour tester les asyncioapplications
  • pytest-cov - bibliothĂšque d'aide pour mesurer la couverture de code par des tests


Ajoutons les lignes suivantes au fichier requirements.txt:



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


Et exécutez dans le terminal:



pip install -r requirements.txt


Ensuite, nous créons Dockerfile. Il décrira le processus de construction et de démarrage de notre démon. Nous allons l'utiliser python:3.8-bustercomme image de base.



Ajoutons les lignes suivantes au fichier 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"]


La derniÚre étape consiste à définir les paramÚtres docker-compose.



Ajoutons les lignes suivantes au fichier docker-compose.yml:



version: "3.7"

services:

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


Tout est prĂȘt. Commençons par construire l'image et vĂ©rifions que l'environnement est correctement configurĂ©.



Exécutons dans le terminal:



docker-compose build


Le processus de construction peut prendre plusieurs minutes. À la fin, vous devriez voir:



Successfully built 5b4ee5e76e35
Successfully tagged monitoring-daemon:latest


Une fois le processus de génération terminé, démarrez le conteneur:



docker-compose up


Tu verras:



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


L'environnement est prĂȘt. Le conteneur commence et se termine par du code 0.



L'Ă©tape suivante consiste Ă  configurer la journalisation et Ă  lire le fichier de configuration.



Journalisation et configuration



Dans cette section, nous allons configurer la journalisation et la lecture du fichier de configuration.



Commençons par ajouter la partie principale de notre application - le conteneur de dépendances (plus loin juste le conteneur). Le conteneur contiendra tous les composants de l'application.



Ajoutons les deux premiers composants. Il s'agit d'un objet de configuration et d'une fonction de configuration de la journalisation.



Éditons 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,
    )


Nous avons utilisé les paramÚtres de configuration avant de définir leurs valeurs. C'est le principe selon lequel le fournisseur fonctionne Configuration.



Nous utilisons d'abord, puis nous définissons les valeurs.



Les paramĂštres de journalisation seront contenus dans le fichier de configuration.



Éditons config.yml:



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


Définissons maintenant une fonction qui lancera notre démon. Elle est généralement appelée main(). Cela créera un conteneur. Le conteneur sera utilisé pour lire le fichier de configuration et appeler la fonction des paramÚtres de journalisation.



Éditons __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()


Le conteneur est le premier objet de l'application. Il est utilisé pour récupérer tous les autres objets.


La journalisation et la lecture de la configuration sont configurées. Dans la section suivante, nous allons créer un gestionnaire de tùches de surveillance.



RĂ©partiteur



Il est temps d'ajouter un gestionnaire de tĂąches de surveillance.



Le répartiteur contiendra une liste des tùches de surveillance et contrÎlera leur exécution. Il exécutera chaque tùche selon le calendrier. Classe Monitor- classe de base pour les tùches de surveillance. Pour créer des tùches spécifiques, vous devez ajouter des classes enfants et implémenter la méthode check().





Ajoutons un répartiteur et une classe de base pour la tùche de surveillance.



Créons dispatcher.pyet monitors.pydans le package monitoringdaemon:



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


Ajoutons les lignes suivantes au fichier 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()


et au dossier 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))


Le rĂ©partiteur doit ĂȘtre ajoutĂ© au conteneur.



Éditons 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
        ),
    )


Chaque composant est ajouté au conteneur.


Enfin, nous devons mettre à jour la fonction main(). Nous allons récupérer le dispatcher du conteneur et appeler sa méthode run().



Éditons __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()


Maintenant, commençons le démon et testons son travail.



Exécutons dans le terminal:



docker-compose up


La sortie devrait ressembler Ă  ceci:



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


Tout fonctionne correctement. Le rĂ©partiteur dĂ©marre et s'arrĂȘte car il n'y a pas de tĂąches de surveillance.



À la fin de cette section, le squelette de notre dĂ©mon est prĂȘt. Dans la section suivante, nous ajouterons la premiĂšre tĂąche de surveillance.



Surveillance example.com



Dans cette section, nous ajouterons une tĂąche de surveillance qui surveillera l'accĂšs Ă  http://example.com .



Nous commencerons par Ă©tendre notre modĂšle de classe avec un nouveau type de tĂąche de surveillance HttpMonitor.



HttpMonitorc'est une classe enfant Monitor. Nous allons implĂ©menter la mĂ©thode check (). Il enverra une requĂȘte HTTP et enregistrera la rĂ©ponse reçue. Les dĂ©tails de la requĂȘte HTTP seront dĂ©lĂ©guĂ©s Ă  la classe HttpClient.





Ajoutons d'abord HttpClient.



Créons un fichier http.pydans un package monitoringdaemon:



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


Et ajoutez-y les lignes suivantes:



"""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


Ensuite, vous devez ajouter HttpClientau conteneur.



Éditons 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
        ),
    )


Nous sommes maintenant prĂȘts Ă  ajouter HttpMonitor. Ajoutons-le au module monitors.



Éditons 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)
        )


Nous sommes tous prĂȘts Ă  ajouter le chĂšque pour http://example.com . Nous devons apporter deux modifications au conteneur:



  • Ajoutez une usine example_monitor.
  • Transfert example_monitorvers le rĂ©partiteur.


Éditons 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,
        ),
    )


Le fournisseur example_monitordépend des valeurs de configuration. Ajoutons ces valeurs:



Edit 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


Tout est prĂȘt. Nous dĂ©marrons le dĂ©mon et vĂ©rifions le travail.



Nous exécutons dans le terminal:



docker-compose up


Et nous voyons une conclusion similaire:



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


Notre démon peut surveiller la disponibilité de l'accÚs à http://example.com .



Ajoutons la surveillance https://httpbin.org .



Surveillance httpbin.org



Dans cette section, nous ajouterons une tĂąche de surveillance qui surveillera l'accĂšs Ă  http://example.com .



L'ajout d'une tĂąche de surveillance pour https://httpbin.org sera plus facile puisque tous les composants sont prĂȘts. Nous devons simplement ajouter un nouveau fournisseur au conteneur et mettre Ă  jour la configuration.



Éditons 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,
        ),
    )


Éditons 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


Lançons le démon et vérifions les journaux.



Exécutons dans le terminal:



docker-compose up


Et nous voyons une conclusion similaire:



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


La partie fonctionnelle est terminée. Le démon surveille la disponibilité de l'accÚs à http://example.com et https://httpbin.org .



Dans la section suivante, nous ajouterons quelques tests.



Des tests



Ce serait bien d'ajouter quelques tests. Faisons cela.



Créez un fichier tests.pydans un package monitoringdaemon:



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


et ajoutez-y les lignes suivantes:



"""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


Pour exécuter les tests, exécutez dans le terminal:



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


Vous devriez obtenir un résultat similaire:



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%


Remarquez comment, dans le test, test_example_monitornous substituons la HttpClientsimulation en utilisant la méthode .override(). De cette façon, vous pouvez remplacer la valeur de retour de n'importe quel fournisseur.



Les mĂȘmes actions sont effectuĂ©es dans le test test_dispatcherpour remplacer les tĂąches de surveillance par des simulacres.





Conclusion



Nous avons construit un démon de surveillance basé sur le asyncioprincipe de l'injection de dépendances. Nous avons utilisé Dependency Injector comme cadre d'injection de dépendances.



L'avantage que vous obtenez avec Dependency Injector est le conteneur.



Le conteneur commence Ă  porter ses fruits lorsque vous devez comprendre ou modifier la structure de votre application. Avec un conteneur, c'est facile car tous les composants de l'application et leurs dĂ©pendances se trouvent au mĂȘme endroit:



"""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,
        ),
    )




Un conteneur comme carte de votre application. Vous savez toujours ce qui dépend de quoi.



Et aprĂšs?






All Articles