برنامج المراقبة الخفي على Asyncio + Dependency Injector - تعليمي لحقن التبعية

مرحبًا ،



أنا مبتكر حاقن التبعية . هذا هو إطار عمل حقن التبعية لبايثون.



هذا برنامج تعليمي آخر لبناء التطبيقات باستخدام حاقن التبعية.



أريد اليوم أن أوضح كيف يمكنك إنشاء برنامج خفي غير متزامن يعتمد على وحدة نمطية asyncio.



يتكون الدليل من الأجزاء التالية:

1. ماذا سنبني؟
2. فحص الأداة
3. هيكل المشروع
4. تهيئة البيئة
5. التسجيل والتكوين
6. المرسل
7. مراقبة example.com
8. مراقبة httpbin.org
9. الاختبارات
10. خاتمة

يمكن العثور على المشروع المكتمل على Github .



للبدء ، من المستحسن أن يكون لديك:

• المعرفة الأولية ب asyncio
• فهم مبدأ حقن التبعية

ماذا سنبني؟

سنقوم ببناء برنامج خفي للمراقبة سيراقب الوصول إلى خدمات الويب.



سيرسل البرنامج الخفي طلبات إلى example.com و httpbin.org كل بضع ثوانٍ. عند تلقي رد ، ستكتب البيانات التالية إلى السجل:

• رمز الاستجابة
• عدد البايتات في الاستجابة
• الوقت المستغرق لإكمال الطلب

فحص الأداة

سنقوم باستخدام عامل الميناء و عامل ميناء-يؤلف . دعنا نتحقق من تثبيتها:

docker --version
docker-compose --version

يجب أن يبدو الناتج كما يلي:

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

إذا لم يتم تثبيت Docker أو Docker-compose ، فيجب تثبيتهما قبل المتابعة. اتبع هذه الأدلة:

• تثبيت Docker
• تركيب عامل البناء

الأدوات جاهزة. دعنا ننتقل إلى هيكل المشروع.

هيكل المشروع

قم بإنشاء مجلد مشروع وانتقل إليه:

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

الآن نحن بحاجة إلى إنشاء هيكل المشروع الأولي. قم بإنشاء ملفات ومجلدات باتباع الهيكل أدناه. ستكون جميع الملفات فارغة في الوقت الحالي. سنملأها لاحقًا.



هيكل المشروع الأولي:

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

الهيكل الأولي للمشروع جاهز. سنقوم بتوسيعه في الأقسام التالية.



بعد ذلك ، ننتظر تهيئة البيئة.

تهيئة البيئة

في هذا القسم ، سنجهز البيئة لبدء برنامجنا الخفي.



تحتاج أولاً إلى تحديد التبعيات. سوف نستخدم حزم مثل هذه:

• dependency-injector - إطار حقن التبعية
• aiohttp - إطار عمل الويب (نحتاج فقط إلى عميل http)
• pyyaml - مكتبة لتحليل ملفات YAML ، وتستخدم لقراءة ملف config
• pytest - إطار الاختبار
• pytest-asyncio- مكتبة مساعدة لاختبار asyncioالتطبيقات
• pytest-cov - مكتبة مساعدة لقياس تغطية الكود بالاختبارات

دعنا نضيف الأسطر التالية إلى الملف requirements.txt:

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

ونفذ في الجهاز:

pip install -r requirements.txt

بعد ذلك ، نخلق Dockerfile. سيصف عملية بناء وبدء برنامجنا الخفي. سوف نستخدمها python:3.8-busterكصورة أساسية.



دعنا نضيف الأسطر التالية إلى الملف 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"]

الخطوة الأخيرة هي تحديد الإعدادات docker-compose.



دعنا نضيف الأسطر التالية إلى الملف docker-compose.yml:

version: "3.7"

services:

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

كل شيء جاهز. لنبدأ في بناء الصورة والتحقق من تهيئة البيئة بشكل صحيح.



لننفذ في المحطة:

docker-compose build

يمكن أن تستغرق عملية الإنشاء عدة دقائق. في النهاية يجب أن ترى:

Successfully built 5b4ee5e76e35
Successfully tagged monitoring-daemon:latest

بعد اكتمال عملية الإنشاء ، ابدأ الحاوية:

docker-compose up

سوف ترى:

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

البيئة جاهزة. تبدأ الحاوية وتنتهي برمز 0.



الخطوة التالية هي إعداد التسجيل وقراءة ملف التكوين.

التسجيل والتكوين

في هذا القسم ، سنقوم بتهيئة التسجيل وقراءة ملف التكوين.



لنبدأ بإضافة الجزء الرئيسي من تطبيقنا - حاوية التبعية (الحاوية فقط). ستحتوي الحاوية على جميع مكونات التطبيق.



دعونا نضيف أول مكونين. هذا هو كائن تكوين ووظيفة لتكوين التسجيل.



دعنا نحرر 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,
    )

استخدمنا معلمات التكوين قبل تحديد قيمها. هذا هو المبدأ الذي يعمل به المزود Configuration.



أولا نستخدم ، ثم نضع القيم.

سيتم تضمين إعدادات التسجيل في ملف التكوين.



دعنا نحرر config.yml:

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

الآن دعونا نحدد وظيفة ستبدأ البرنامج الخفي لدينا. عادة ما يتم استدعاؤها main(). سيخلق حاوية. سيتم استخدام الحاوية لقراءة ملف التكوين واستدعاء وظيفة إعدادات التسجيل.



دعنا نحرر __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()

الحاوية هي الكائن الأول في التطبيق. يتم استخدامه للحصول على جميع الأشياء الأخرى.

تكوين تسجيل وقراءة تكوين. في القسم التالي ، سننشئ مدير مهام المراقبة.

المرسل

حان الوقت لإضافة مدير مهام المراقبة.



سيحتوي المرسل على قائمة مهام المراقبة والتحكم في تنفيذها. سوف يؤدي كل مهمة وفقًا للجدول الزمني. فئة Monitor- فئة أساسية لمهام المراقبة. لإنشاء مهام محددة ، تحتاج إلى إضافة فئات فرعية وتنفيذ الطريقة check().





دعنا نضيف مرسلًا وفئة أساسية لمهمة المراقبة.



لنقم بإنشاء dispatcher.pyوفي monitors.pyالحزمة monitoringdaemon:

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

دعنا نضيف الأسطر التالية إلى الملف 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()

وإلى الملف 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))

يجب إضافة المرسل إلى الحاوية.



دعنا نحرر 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
        ),
    )

يضاف كل مكون إلى الحاوية.

أخيرًا ، نحتاج إلى تحديث الوظيفة main(). سوف نحصل على المرسل من الحاوية وندعو أسلوبه run().



دعنا نحرر __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()

الآن لنبدأ البرنامج الخفي ونختبر عمله.



لننفذ في المحطة:

docker-compose up

يجب أن يبدو الإخراج كما يلي:

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

كل شيء يعمل بشكل صحيح. يبدأ المرسل ويتوقف نظرًا لعدم وجود مهام مراقبة.



بنهاية هذا القسم ، يكون الهيكل العظمي لشيطاننا جاهزًا. في القسم التالي ، سنضيف مهمة المراقبة الأولى.

مراقبة example.com

في هذا القسم ، سنضيف مهمة مراقبة ستراقب الوصول إلى http://example.com .



سنبدأ بتوسيع نموذج الفصل الخاص بنا بنوع جديد من مهام المراقبة HttpMonitor.



HttpMonitorإنها فئة طفل Monitor. سنقوم بتنفيذ طريقة الشيك (). سيرسل طلب HTTP ويسجل الاستجابة المستلمة. سيتم تفويض تفاصيل طلب HTTP إلى الفصل الدراسي HttpClient.





دعنا نضيف أولا HttpClient.



لنقم بإنشاء ملف http.pyفي حزمة monitoringdaemon:

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

وأضف إليها الأسطر التالية:

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

بعد ذلك ، تحتاج إلى الإضافة HttpClientإلى الحاوية.



دعنا نحرر 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
        ),
    )

نحن الآن جاهزون للإضافة HttpMonitor. دعنا نضيفه إلى الوحدة monitors.



دعنا نحرر 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)
        )

نحن مستعدون جميعًا لإضافة التحقق من http://example.com . نحتاج إلى إجراء تغييرين على الحاوية:

• أضف مصنعًا example_monitor.
• نقل example_monitorإلى المرسل.

دعنا نحرر 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,
        ),
    )

يعتمد الموفر example_monitorعلى قيم التكوين. دعونا إضافة هذه القيم:



تحرير 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

كل شيء جاهز. نبدأ البرنامج الخفي ونتحقق من العمل.



ننفذ في المحطة:

docker-compose up

ونرى نتيجة مماثلة:

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

يمكن للبرنامج الخفي لدينا مراقبة مدى توفر الوصول إلى http://example.com .



دعنا نضيف مراقبة https://httpbin.org .

مراقبة httpbin.org

في هذا القسم ، سنضيف مهمة مراقبة ستراقب الوصول إلى http://example.com . ستكون



إضافة مهمة مراقبة لـ https://httpbin.org أسهل لأن جميع المكونات جاهزة. نحتاج فقط إلى إضافة مزود جديد إلى الحاوية وتحديث التكوين.



دعنا نحرر 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,
        ),
    )

دعنا نحرر 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

لنبدأ البرنامج الخفي ونتحقق من السجلات.



لننفذ في المحطة:

docker-compose up

ونرى نتيجة مماثلة:

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

اكتمل الجزء الوظيفي. يراقب البرنامج الخفي توفر الوصول إلى http://example.com و https://httpbin.org .



في القسم التالي ، سنضيف بعض الاختبارات.

الاختبارات

سيكون من الجيد إضافة بعض الاختبارات. لنفعل ذلك.



قم بإنشاء ملف tests.pyفي حزمة monitoringdaemon:

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

وأضف إليها الأسطر التالية:

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

لإجراء الاختبارات ، قم بتشغيل الجهاز:

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

يجب أن تحصل على نتيجة مماثلة:

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%

لاحظ كيف في الاختبار test_example_monitorاستبدلنا HttpClientهمية باستخدام طريقة .override(). بهذه الطريقة ، يمكنك تجاوز القيمة المرجعة لأي مزود.



يتم تنفيذ نفس الإجراءات في الاختبار test_dispatcherلاستبدال مهام المراقبة بالنماذج.

خاتمة

لقد أنشأنا برنامجًا خفيًا للمراقبة على أساس asyncioمبدأ حقن التبعية. استخدمنا حاقن التبعية كإطار لحقن التبعية.



الميزة التي تحصل عليها مع Dependency Injector هي الحاوية.



تبدأ الحاوية في الدفع عندما تحتاج إلى فهم أو تغيير هيكل التطبيق الخاص بك. باستخدام الحاوية ، يكون الأمر سهلاً لأن جميع مكونات التطبيق وتبعياتها موجودة في مكان واحد:

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

الحاوية كخريطة لتطبيقك. أنت تعرف دائمًا ما يعتمد على ما.

ماذا بعد؟

• تعرف على المزيد حول Dependency Injector على GitHub
• تحقق من الوثائق في قراءة المستندات
• لديك سؤال أو تجد علة؟ افتح مشكلة على جيثب