Memantau daemon di Asyncio + Dependency Injector - panduan injeksi ketergantungan

Hai,



saya pencipta Dependency Injector . Ini adalah kerangka kerja injeksi ketergantungan untuk Python.



Ini adalah tutorial lain untuk membangun aplikasi dengan Dependency Injector.



Hari ini saya ingin menunjukkan bagaimana Anda dapat membangun daemon asinkron berdasarkan modul asyncio.



Manual terdiri dari bagian-bagian berikut:

1. Apa yang akan kita bangun?
2. Pemeriksaan alat
3. Struktur proyek
4. Mempersiapkan lingkungan
5. Logging dan konfigurasi
6. Operator
7. Memantau example.com
8. Memantau httpbin.org
9. Tes
10. Kesimpulan

Proyek yang telah selesai dapat ditemukan di Github .



Untuk memulai, diinginkan untuk memiliki:

• Pengetahuan awal tentang asyncio
• Memahami prinsip injeksi ketergantungan

Apa yang akan kita bangun?

Kami akan membangun daemon pemantauan yang akan memantau akses ke layanan web.



Daemon akan mengirimkan permintaan ke example.com dan httpbin.org setiap beberapa detik. Saat menerima respons, itu akan menulis data berikut ke log:

• Kode respon
• Jumlah byte sebagai tanggapan
• Waktu yang dibutuhkan untuk menyelesaikan permintaan

Pemeriksaan alat

Kami akan menggunakan Docker dan docker-compose . Mari kita periksa apakah sudah terpasang:

docker --version
docker-compose --version

Outputnya akan terlihat seperti ini:

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

Jika Docker atau docker-compose tidak diinstal, mereka harus diinstal sebelum melanjutkan. Ikuti panduan ini:

• Menginstal Docker
• Memasang docker-compose

Alat sudah siap. Mari beralih ke struktur proyek.

Struktur proyek

Buat folder proyek dan buka:

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

Sekarang kita perlu membuat struktur proyek awal. Buat file dan folder mengikuti struktur di bawah ini. Semua file akan kosong untuk saat ini. Kami akan mengisinya nanti.



Struktur proyek awal:

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

Struktur proyek awal sudah siap. Kami akan mengembangkannya di bagian berikut.



Selanjutnya kita tunggu persiapan lingkungan.

Mempersiapkan lingkungan

Di bagian ini, kami akan menyiapkan lingkungan untuk memulai daemon kami.



Pertama, Anda perlu mendefinisikan dependensi. Kami akan menggunakan paket seperti ini:

• dependency-injector - kerangka injeksi ketergantungan
• aiohttp - kerangka web (kami hanya membutuhkan klien http)
• pyyaml - perpustakaan untuk mem-parsing file YAML, digunakan untuk membaca konfigurasi
• pytest - kerangka pengujian
• pytest-asyncio- perpustakaan pembantu untuk menguji asyncioaplikasi
• pytest-cov - Perpustakaan pembantu untuk mengukur cakupan kode dengan tes

Mari tambahkan baris berikut ke file requirements.txt:

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

Dan jalankan di terminal:

pip install -r requirements.txt

Selanjutnya kita buat Dockerfile. Ini akan menjelaskan proses membangun dan memulai daemon kita. Kami akan menggunakannya python:3.8-bustersebagai gambar dasar.



Mari tambahkan baris berikut ke file 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"]

Langkah terakhir adalah menentukan pengaturan docker-compose.



Mari tambahkan baris berikut ke file docker-compose.yml:

version: "3.7"

services:

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

Semuanya sudah siap. Mari mulai membuat gambar dan periksa apakah lingkungan telah dikonfigurasi dengan benar.



Mari kita jalankan di terminal:

docker-compose build

Proses pembuatan dapat memakan waktu beberapa menit. Pada akhirnya, Anda akan melihat:

Successfully built 5b4ee5e76e35
Successfully tagged monitoring-daemon:latest

Setelah proses build selesai, mulai penampung:

docker-compose up

Kamu akan lihat:

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

Lingkungan sudah siap. Penampung dimulai dan keluar dengan kode 0.



Langkah selanjutnya adalah mengatur logging dan membaca file konfigurasi.

Logging dan konfigurasi

Di bagian ini, kami akan mengkonfigurasi logging dan membaca file konfigurasi.



Mari kita mulai dengan menambahkan bagian utama dari aplikasi kita - wadah ketergantungan (selanjutnya hanya wadah). Penampung akan berisi semua komponen aplikasi.



Mari tambahkan dua komponen pertama. Ini adalah objek konfigurasi dan fungsi untuk mengkonfigurasi logging.



Mari edit 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,
    )

Kami menggunakan parameter konfigurasi sebelum menetapkan nilainya. Ini adalah prinsip yang digunakan oleh penyedia Configuration.



Pertama kita gunakan, lalu kita atur nilainya.

Pengaturan pencatatan akan terdapat di file konfigurasi.



Mari edit config.yml:

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

Sekarang mari kita definisikan fungsi yang akan memulai daemon kita. Dia biasanya dipanggil main(). Ini akan membuat wadah. Kontainer akan digunakan untuk membaca file konfigurasi dan memanggil fungsi pengaturan logging.



Mari edit __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()

Kontainer adalah objek pertama dalam aplikasi. Ini digunakan untuk mendapatkan semua objek lainnya.

Log konfigurasi dan pembacaan dikonfigurasi. Di bagian selanjutnya, kami akan membuat pengelola tugas pemantauan.

Operator

Saatnya menambahkan pengelola tugas pemantauan.



Petugas operator akan berisi daftar tugas pemantauan dan kontrol pelaksanaannya. Ia akan melaksanakan setiap tugas sesuai jadwal. Kelas Monitor- kelas dasar untuk tugas pemantauan. Untuk membuat tugas tertentu, Anda perlu menambahkan kelas anak dan mengimplementasikan metode tersebut check().





Mari tambahkan dispatcher dan kelas dasar untuk tugas pemantauan.



Mari buat dispatcher.pydan monitors.pydi dalam paket monitoringdaemon:

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

Mari tambahkan baris berikut ke file 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()

dan ke file 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))

Petugas operator perlu ditambahkan ke penampung.



Mari edit 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
        ),
    )

Setiap komponen ditambahkan ke wadah.

Terakhir, kita perlu memperbarui fungsinya main(). Kita akan mendapatkan dispatcher dari container dan memanggil metodenya run().



Mari edit __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()

Sekarang mari kita mulai daemon dan menguji kerjanya.



Mari kita jalankan di terminal:

docker-compose up

Outputnya akan terlihat seperti ini:

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

Semuanya bekerja dengan benar. Petugas operator memulai dan berhenti karena tidak ada tugas pemantauan.



Pada akhir bagian ini, kerangka iblis kita sudah siap. Di bagian selanjutnya, kami akan menambahkan tugas pemantauan pertama.

Memantau example.com

Di bagian ini, kami akan menambahkan tugas pemantauan yang akan memantau akses ke http://example.com .



Kami akan mulai dengan memperluas model kelas kami dengan jenis tugas pemantauan baru HttpMonitor.



HttpMonitoritu adalah kelas anak-anak Monitor. Kami akan menerapkan metode check (). Ini akan mengirim permintaan HTTP dan mencatat respons yang diterima. Detail permintaan HTTP akan didelegasikan ke kelas HttpClient.





Mari tambahkan dulu HttpClient.



Mari buat file http.pydalam sebuah paket monitoringdaemon:

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

Dan tambahkan baris berikut ke dalamnya:

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

Selanjutnya, Anda perlu menambahkan HttpClientke wadah.



Mari edit 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
        ),
    )

Kami sekarang siap untuk menambahkan HttpMonitor. Mari tambahkan ke modul monitors.



Mari edit 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)
        )

Kami siap untuk menambahkan cek untuk http://example.com . Kita perlu membuat dua perubahan pada penampung:

• Tambahkan pabrik example_monitor.
• Transfer example_monitorke petugas operator.

Mari edit 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,
        ),
    )

Penyedia example_monitorbergantung pada nilai konfigurasi. Mari tambahkan nilai-nilai ini:



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

Semuanya sudah siap. Kami memulai daemon dan memeriksa pekerjaan.



Kami mengeksekusi di terminal:

docker-compose up

Dan kami melihat kesimpulan serupa:

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

Daemon kami dapat memantau ketersediaan akses ke http://example.com .



Mari tambahkan pemantauan https://httpbin.org .

Memantau httpbin.org

Di bagian ini, kami akan menambahkan tugas pemantauan yang akan memantau akses ke http://example.com .



Menambahkan tugas pemantauan untuk https://httpbin.org akan lebih mudah karena semua komponen sudah siap. Kami hanya perlu menambahkan penyedia baru ke penampung dan memperbarui konfigurasi.



Mari edit 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,
        ),
    )

Mari 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

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

Mari mulai daemon dan periksa lognya.



Mari kita jalankan di terminal:

docker-compose up

Dan kami melihat kesimpulan serupa:

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

Bagian fungsional selesai. Daemon memantau ketersediaan akses ke http://example.com dan https://httpbin.org .



Di bagian selanjutnya, kami akan menambahkan beberapa tes.

Tes

Akan menyenangkan menambahkan beberapa tes. Ayo lakukan itu.



Buat file tests.pydalam sebuah paket monitoringdaemon:

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

dan tambahkan baris berikut ke dalamnya:

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

Untuk menjalankan tes, jalankan di terminal:

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

Anda harus mendapatkan hasil yang serupa:

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%

Perhatikan bagaimana dalam pengujian test_example_monitorkami mengganti HttpClientmock menggunakan metode .override(). Dengan cara ini, Anda bisa mengganti nilai kembalian dari penyedia mana pun.



Tindakan yang sama dilakukan dalam pengujian test_dispatcheruntuk menggantikan tugas pemantauan dengan tiruan.

Kesimpulan

Kami membangun daemon pemantauan berdasarkan asyncioprinsip injeksi ketergantungan. Kami menggunakan Dependency Injector sebagai kerangka kerja injeksi ketergantungan.



Keuntungan yang Anda dapatkan dengan Dependency Injector adalah wadah.



Penampung mulai terbayar saat Anda perlu memahami atau mengubah struktur aplikasi Anda. Dengan container, mudah karena semua komponen aplikasi dan dependensinya berada di satu tempat:

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

Sebuah wadah sebagai peta aplikasi Anda. Anda selalu tahu apa yang tergantung pada apa.

Apa berikutnya?

• Pelajari lebih lanjut tentang Dependency Injector di GitHub
• Lihat dokumentasinya di Read the Docs
• Punya pertanyaan atau temukan bug? Buka masalah di Github