💅🏻 ↪️ 👋🏿 Monitoreo del demonio en Asyncio + Dependency Injector - tutorial de inyección de dependencias 🤫 🎒 ♌️

Hola,

soy el creador de Dependency Injector . Este es un marco de inyección de dependencia para Python.

Este es otro tutorial para crear aplicaciones con Dependency Injector.

Hoy quiero mostrar cómo se puede construir un demonio asincrónico basado en un módulo asyncio.

El manual consta de las siguientes partes:

El proyecto completo se puede encontrar en Github .

Para empezar, es deseable tener:

Conocimiento inicial de asyncio
Comprender el principio de inyección de dependencia

¿Qué vamos a construir?

Construiremos un demonio de monitoreo que monitoreará el acceso a los servicios web.

El demonio enviará solicitudes a example.com y httpbin.org cada pocos segundos. Al recibir una respuesta, escribirá los siguientes datos en el registro:

Código de respuesta
Número de bytes en respuesta
Tiempo necesario para completar la solicitud

Comprobación de herramientas

Usaremos Docker y docker-compose . Comprobemos que estén instalados:

docker --version
docker-compose --version

La salida debería verse así:

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

Si Docker o docker-compose no están instalados, es necesario instalarlos antes de continuar. Siga estas guías:

Instalación de Docker

Instalación de docker-compose

Las herramientas están listas. Pasemos a la estructura del proyecto.

Estructura del proyecto

Crea una carpeta de proyecto y ve a ella:

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

Ahora necesitamos crear una estructura de proyecto inicial. Cree archivos y carpetas siguiendo la estructura siguiente. Todos los archivos estarán vacíos por ahora. Los llenaremos más tarde.

Estructura inicial del proyecto:

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

La estructura del proyecto inicial está lista. Lo ampliaremos en las siguientes secciones.

A continuación, estamos a la espera de la preparación del entorno.

Preparando el medio ambiente

En esta sección, prepararemos el entorno para iniciar nuestro demonio.

Primero necesitas definir dependencias. Usaremos paquetes como este:

dependency-injector - marco de inyección de dependencia
aiohttp - marco web (solo necesitamos un cliente http)
pyyaml - biblioteca para analizar archivos YAML, utilizada para leer la configuración
pytest - marco de prueba
pytest-asyncio- biblioteca auxiliar para probar asyncioaplicaciones
pytest-cov - biblioteca auxiliar para medir la cobertura del código mediante pruebas

Agreguemos las siguientes líneas al archivo requirements.txt:

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

Y ejecutar en la terminal:

pip install -r requirements.txt

A continuación, creamos Dockerfile. Describirá el proceso de construcción e inicio de nuestro demonio. Lo usaremos python:3.8-bustercomo imagen base.

Agreguemos las siguientes líneas al archivo 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"]

El último paso es definir la configuración docker-compose.

Agreguemos las siguientes líneas al archivo docker-compose.yml:

version: "3.7"

services:

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

Todo está listo. Comencemos a construir la imagen y verifiquemos que el entorno esté configurado correctamente.

Ejecutemos en la terminal:

docker-compose build

El proceso de construcción puede tardar varios minutos. Al final, debería ver:

Successfully built 5b4ee5e76e35
Successfully tagged monitoring-daemon:latest

Una vez completado el proceso de compilación, inicie el contenedor:

docker-compose up

Ya verás:

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

El ambiente está listo. El contenedor comienza y termina con código 0.

El siguiente paso es configurar el registro y leer el archivo de configuración.

Registro y configuración

En esta sección, configuraremos el registro y la lectura del archivo de configuración.

Comencemos agregando la parte principal de nuestra aplicación: el contenedor de dependencias (además, solo el contenedor). El contenedor contendrá todos los componentes de la aplicación.

Agreguemos los dos primeros componentes. Este es un objeto de configuración y una función para configurar el registro.

Editemos 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 los parámetros de configuración antes de establecer sus valores. Este es el principio por el que trabaja el proveedor Configuration.

Primero usamos, luego establecemos los valores.

La configuración de registro estará contenida en el archivo de configuración.

Editemos config.yml:

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

Ahora definamos una función que iniciará nuestro demonio. Generalmente la llaman main(). Creará un contenedor. El contenedor se utilizará para leer el archivo de configuración y llamar a la función de configuración de registro.

Editemos __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()

El contenedor es el primer objeto de la aplicación. Se usa para obtener todos los demás objetos.

El registro y la lectura de la configuración están configurados. En la siguiente sección, crearemos un administrador de tareas de monitoreo.

Despachador

Es hora de agregar un administrador de tareas de monitoreo.

El despachador contendrá una lista de tareas de monitoreo y controlará su ejecución. Realizará cada tarea de acuerdo con el horario. Clase Monitor- clase base para tareas de monitoreo. Para crear tareas específicas, debe agregar clases secundarias e implementar el método check().

Agreguemos un despachador y una clase base para la tarea de monitoreo.

Creemos dispatcher.pyy monitors.pyen el paquete monitoringdaemon:

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

Agreguemos las siguientes líneas al archivo 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()

y al archivo 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))

El despachador debe agregarse al contenedor.

Editemos 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 se agrega al contenedor.

Finalmente, necesitamos actualizar la función main(). Obtendremos el despachador del contenedor y llamaremos a su método run().

Editemos __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()

Ahora iniciemos el demonio y probemos su funcionamiento.

Ejecutemos en la terminal:

docker-compose up

La salida debería verse así:

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

Todo funciona correctamente. El despachador se inicia y se detiene ya que no hay tareas de supervisión.

Al final de esta sección, el esqueleto de nuestro demonio está listo. En la siguiente sección, agregaremos la primera tarea de monitoreo.

Monitoreo example.com

En esta sección, agregaremos una tarea de monitoreo que monitoreará el acceso a http://example.com .

Comenzaremos ampliando nuestro modelo de clase con un nuevo tipo de tarea de supervisión HttpMonitor.

HttpMonitores una clase infantil Monitor. Implementaremos el método check (). Enviará una solicitud HTTP y registrará la respuesta recibida. Los detalles de la solicitud HTTP se delegarán a la clase HttpClient.

Agreguemos primero HttpClient.

Creemos un archivo http.pyen un paquete monitoringdaemon:

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

Y agregue las siguientes líneas:

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

A continuación, debe agregar HttpCliental contenedor.

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

Ahora estamos listos para agregar HttpMonitor. Agreguémoslo al módulo monitors.

Editemos 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 listos para agregar el cheque para http://example.com . Necesitamos hacer dos cambios en el contenedor:

Agrega una fábrica example_monitor.
Traslado example_monitoral despachador.

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

El proveedor example_monitordepende de los valores de configuración. Agreguemos estos 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

Todo está listo. Iniciamos el demonio y comprobamos el trabajo.

Ejecutamos en la terminal:

docker-compose up

Y vemos una conclusión similar:

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

Nuestro demonio puede monitorear la disponibilidad de acceso a http://example.com .

Agreguemos monitoreo https://httpbin.org .

Monitoreo de httpbin.org

En esta sección, agregaremos una tarea de monitoreo que monitoreará el acceso a http://example.com .

Agregar una tarea de monitoreo para https://httpbin.org será más fácil porque todos los componentes están listos. Solo necesitamos agregar un nuevo proveedor al contenedor y actualizar la configuración.

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

Editemos 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

Iniciemos el demonio y revisemos los registros.

Ejecutemos en la terminal:

docker-compose up

Y vemos una conclusión similar:

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

Se completa la parte funcional. El demonio monitorea la disponibilidad de acceso a http://example.com y https://httpbin.org .

En la siguiente sección, agregaremos algunas pruebas.

Pruebas

Sería bueno agregar algunas pruebas. Vamos a hacer eso.

Cree un archivo tests.pyen un paquete monitoringdaemon:

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

y agregue las siguientes líneas:

"""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 ejecutar las pruebas, ejecute en la terminal:

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

Debería obtener un resultado similar:

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 cómo en la prueba test_example_monitorsustituimos el HttpClientsimulacro por el método .override(). De esta forma, puede anular el valor de retorno de cualquier proveedor.

Las mismas acciones se realizan en la prueba test_dispatcherpara reemplazar las tareas de monitoreo con simulacros.

Conclusión

Creamos un demonio de monitoreo basado en el asyncioprincipio de inyección de dependencia. Usamos Dependency Injector como un marco de inyección de dependencias.

La ventaja que obtiene con Dependency Injector es el contenedor.

El contenedor comienza a dar sus frutos cuando necesita comprender o cambiar la estructura de su aplicación. Con un contenedor, es fácil porque todos los componentes de la aplicación y sus dependencias están en un solo 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,
        ),
    )

Un contenedor como mapa de su aplicación. Siempre sabes qué depende de qué.

¿Que sigue?

Obtenga más información sobre el inyector de dependencia en GitHub
Consulte la documentación en Leer los documentos
¿Tiene alguna pregunta o encuentra un error? Abrir un problema en Github

Monitoreo del demonio en Asyncio + Dependency Injector - tutorial de inyección de dependencias