Clever Geek Handbook

Aiohttp + حاقن التبعية - تعليمي لحقن التبعية

مرحبًا ،



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



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



في هذا البرنامج التعليمي ، أود أن أوضح لك كيفية استخدام حاقن التبعية لتطوير aiohttpالتطبيق.



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



  1. ماذا سنبني؟
  2. تهيئة البيئة
  3. هيكل المشروع
  4. تثبيت التبعيات
  5. تطبيق الحد الأدنى
  6. عميل Giphy API
  7. خدمة البحث
  8. ربط البحث
  9. قليلا من إعادة بناء ديون
  10. إضافة الاختبارات
  11. خاتمة


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



للبدء ، يجب أن يكون لديك:



  • Python 3.5+
  • بيئة افتراضية


ويستحب أن يكون لديك:



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


ماذا سنبني؟







سنقوم ببناء تطبيق REST API الذي يبحث عن صور متحركة مضحكة على Giphy . دعنا نسميها Giphy Navigator.



كيف يعمل Giphy Navigator؟



  • يرسل العميل طلبًا يوضح ما الذي تبحث عنه وعدد النتائج التي يجب إرجاعها.
  • يعرض Giphy Navigator استجابة json.
  • الجواب يشمل:

    • استعلام بحث
    • عدد النتائج
    • قائمة عناوين URL بتنسيق GIF


استجابة العينة: 



{
    "query": "Dependency Injector",
    "limit": 10,
    "gifs": [
        {
            "url": "https://giphy.com/gifs/boxes-dependent-swbf2-6Eo7KzABxgJMY"
        },
        {
            "url": "https://giphy.com/gifs/depends-J56qCcOhk6hKE"
        },
        {
            "url": "https://giphy.com/gifs/web-series-ccstudios-bro-dependent-1lhU8KAVwmVVu"
        },
        {
            "url": "https://giphy.com/gifs/TheBoysTV-friends-friend-weneedeachother-XxR9qcIwcf5Jq404Sx"
        },
        {
            "url": "https://giphy.com/gifs/netflix-a-series-of-unfortunate-events-asoue-9rgeQXbwoK53pcxn7f"
        },
        {
            "url": "https://giphy.com/gifs/black-and-white-sad-skins-Hs4YzLs2zJuLu"
        },
        {
            "url": "https://giphy.com/gifs/always-there-for-you-i-am-here-PlayjhCco9jHBYrd9w"
        },
        {
            "url": "https://giphy.com/gifs/stream-famous-dollar-YT2dvOByEwXCdoYiA1"
        },
        {
            "url": "https://giphy.com/gifs/i-love-you-there-for-am-1BhGzgpZXYWwWMAGB1"
        },
        {
            "url": "https://giphy.com/gifs/life-like-twerk-9hlnWxjHqmH28"
        }
    ]
}


جهز البيئة



لنبدأ بإعداد البيئة.



بادئ ذي بدء ، نحتاج إلى إنشاء مجلد مشروع وبيئة افتراضية: 



mkdir giphynav-aiohttp-tutorial
cd giphynav-aiohttp-tutorial
python3 -m venv venv


لنقم الآن بتنشيط البيئة الافتراضية: 



. venv/bin/activate


البيئة جاهزة ، لنبدأ الآن بهيكل المشروع.



هيكل المشروع



في هذا القسم ، سننظم هيكل المشروع.



لنقم بإنشاء الهيكل التالي في المجلد الحالي. اترك كل الملفات فارغة الآن.



الهيكل الأولي: 



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   └── views.py
├── venv/
└── requirements.txt


تثبيت التبعيات



حان الوقت لتثبيت التبعيات. سوف نستخدم حزم مثل هذه:



  • dependency-injector - إطار حقن التبعية
  • aiohttp - إطار عمل الويب
  • aiohttp-devtools - مكتبة مساعدة توفر خادمًا لتطوير إعادة التشغيل المباشر
  • pyyaml - مكتبة لتحليل ملفات YAML ، وتستخدم لقراءة ملف config
  • pytest-aiohttp- مكتبة مساعدة لاختبار aiohttpالتطبيقات
  • pytest-cov - مكتبة مساعدة لقياس تغطية الكود بالاختبارات


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



dependency-injector
aiohttp
aiohttp-devtools
pyyaml
pytest-aiohttp
pytest-cov


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



pip install -r requirements.txt


تثبيت بالإضافة إلى ذلك httpie. إنه عميل HTTP سطر أوامر. وسوف

استخدامه لاختبار API يدويا.



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



pip install httpie


تم تثبيت التبعيات. لنقم الآن ببناء تطبيق بسيط.



تطبيق الحد الأدنى



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



دعنا نحرر views.py: 



"""Views module."""

from aiohttp import web


async def index(request: web.Request) -> web.Response:
    query = request.query.get('query', 'Dependency Injector')
    limit = int(request.query.get('limit', 10))

    gifs = []

    return web.json_response(
        {
            'query': query,
            'limit': limit,
            'gifs': gifs,
        },
    )


الآن دعنا نضيف حاوية للاعتماديات (يشار إليها فيما يلي فقط بالحاوية). ستحتوي الحاوية على جميع مكونات التطبيق. دعونا نضيف أول مكونين. هذا aiohttpتطبيق وعرض index.



دعنا نحرر containers.py: 



"""Application containers module."""

from dependency_injector import containers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import views


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

    app = aiohttp.Application(web.Application)

    index_view = aiohttp.View(views.index)


الآن نحن بحاجة إلى إنشاء مصنع aiohttpتطبيقات. عادة ما يطلق عليه

create_app(). سيخلق حاوية. سيتم استخدام الحاوية لإنشاء aiohttpالتطبيق. الخطوة الأخيرة هي إعداد التوجيه - سنقوم بتعيين طريقة عرض index_viewمن الحاوية لمعالجة الطلبات إلى جذر "/"تطبيقنا.



دعنا نحرر application.py: 



"""Application module."""

from aiohttp import web

from .containers import ApplicationContainer


def create_app():
    """Create and return aiohttp application."""
    container = ApplicationContainer()

    app: web.Application = container.app()
    app.container = container

    app.add_routes([
        web.get('/', container.index_view.as_view()),
    ])

    return app


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


نحن الآن جاهزون لإطلاق تطبيقنا: قم



بتشغيل الأمر في المحطة: 



adev runserver giphynavigator/application.py --livereload


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



[18:52:59] Starting aux server at http://localhost:8001 ◆
[18:52:59] Starting dev server at http://localhost:8000 ●


نستخدمها httpieللتحقق من تشغيل الخادم: 



http http://127.0.0.1:8000/


سوف ترى: 



HTTP/1.1 200 OK
Content-Length: 844
Content-Type: application/json; charset=utf-8
Date: Wed, 29 Jul 2020 21:01:50 GMT
Server: Python/3.8 aiohttp/3.6.2

{
    "gifs": [],
    "limit": 10,
    "query": "Dependency Injector"
}


تطبيق الحد الأدنى جاهز. لنقم بتوصيل واجهة برمجة تطبيقات Giphy.



عميل Giphy API



في هذا القسم ، سنقوم بدمج تطبيقنا مع Giphy API. سننشئ عميل API الخاص بنا باستخدام جانب العميل aiohttp.



قم بإنشاء ملف فارغ giphy.pyفي الحزمة giphynavigator: 



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── giphy.py
│   └── views.py
├── venv/
└── requirements.txt


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



"""Giphy client module."""

from aiohttp import ClientSession, ClientTimeout


class GiphyClient:

    API_URL = 'http://api.giphy.com/v1'

    def __init__(self, api_key, timeout):
        self._api_key = api_key
        self._timeout = ClientTimeout(timeout)

    async def search(self, query, limit):
        """Make search API call and return result."""
        if not query:
            return []

        url = f'{self.API_URL}/gifs/search'
        params = {
            'q': query,
            'api_key': self._api_key,
            'limit': limit,
        }
        async with ClientSession(timeout=self._timeout) as session:
            async with session.get(url, params=params) as response:
                if response.status != 200:
                    response.raise_for_status()
                return await response.json()


الآن نحن بحاجة إلى إضافة GiphyClient إلى الحاوية. يحتوي GiphyClient على تبعيتين يجب تمريرهما عند إنشائه: مفتاح API ومهلة الطلب. للقيام بذلك ، سنحتاج إلى استخدام مزودين جديدين من الوحدة dependency_injector.providers:



  • سيقوم الموفر Factoryبإنشاء GiphyClient.
  • Configurationسيرسل الموفر مفتاح API والمهلة إلى GiphyClient.


دعنا نحرر containers.py: 



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, views


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

    app = aiohttp.Application(web.Application)

    config = providers.Configuration()

    giphy_client = providers.Factory(
        giphy.GiphyClient,
        api_key=config.giphy.api_key,
        timeout=config.giphy.request_timeout,
    )

    index_view = aiohttp.View(views.index)


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



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



الآن دعنا نضيف ملف التكوين.

سوف نستخدم YAML.



قم بإنشاء ملف فارغ config.ymlفي جذر المشروع: 



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── giphy.py
│   └── views.py
├── venv/
├── config.yml
└── requirements.txt


واملأه بالأسطر التالية: 



giphy:
  request_timeout: 10


سنستخدم متغير بيئة لتمرير مفتاح API GIPHY_API_KEY .



نحتاج الآن إلى التعديل create_app()للقيام بإجراءين عند بدء التطبيق:



  • تكوين التحميل من config.yml
  • تحميل مفتاح API من متغير البيئة GIPHY_API_KEY


تحرير application.py: 



"""Application module."""

from aiohttp import web

from .containers import ApplicationContainer


def create_app():
    """Create and return aiohttp application."""
    container = ApplicationContainer()
    container.config.from_yaml('config.yml')
    container.config.giphy.api_key.from_env('GIPHY_API_KEY')

    app: web.Application = container.app()
    app.container = container

    app.add_routes([
        web.get('/', container.index_view.as_view()),
    ])

    return app


نحتاج الآن إلى إنشاء مفتاح API وتعيينه على متغير بيئة.



لكي لا تضيع الوقت في هذا ، استخدم الآن هذا المفتاح: 



export GIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0


اتبع هذا البرنامج التعليمي لإنشاء مفتاح Giphy API الخاص بك .


اكتمل إنشاء وتهيئة عميل Giphy API. دعنا ننتقل إلى خدمة البحث.



خدمة البحث



حان الوقت لإضافة خدمة البحث SearchService. هو سوف:



  • بحث
  • تلقى تنسيق الاستجابة


SearchServiceسوف تستخدم GiphyClient.



قم بإنشاء ملف فارغ services.pyفي الحزمة giphynavigator: 



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── giphy.py
│   ├── services.py
│   └── views.py
├── venv/
└── requirements.txt


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



"""Services module."""

from .giphy import GiphyClient


class SearchService:

    def __init__(self, giphy_client: GiphyClient):
        self._giphy_client = giphy_client

    async def search(self, query, limit):
        """Search for gifs and return formatted data."""
        if not query:
            return []

        result = await self._giphy_client.search(query, limit)

        return [{'url': gif['url']} for gif in result['data']]


عند الإنشاء ، SearchServiceتحتاج إلى النقل GiphyClient. سنشير إلى هذا عندما نضيفه SearchServiceإلى الحاوية.



دعنا نحرر containers.py: 



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, services, views


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

    app = aiohttp.Application(web.Application)

    config = providers.Configuration()

    giphy_client = providers.Factory(
        giphy.GiphyClient,
        api_key=config.giphy.api_key,
        timeout=config.giphy.request_timeout,
    )

    search_service = providers.Factory(
        services.SearchService,
        giphy_client=giphy_client,
    )

    index_view = aiohttp.View(views.index)


خدمة البحث SearchServiceاكتملت الآن . في القسم التالي ، سنربطها برأينا.



ربط البحث



نحن الآن جاهزون للبحث للعمل. دعونا نستخدم SearchServiceفي indexالعرض.



تحرير views.py: 



"""Views module."""

from aiohttp import web

from .services import SearchService


async def index(
        request: web.Request,
        search_service: SearchService,
) -> web.Response:
    query = request.query.get('query', 'Dependency Injector')
    limit = int(request.query.get('limit', 10))

    gifs = await search_service.search(query, limit)

    return web.json_response(
        {
            'query': query,
            'limit': limit,
            'gifs': gifs,
        },
    )


لنقم الآن بتغيير الحاوية لتمرير التبعية SearchServiceإلى العرض indexعند استدعائها.



تحرير containers.py: 



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, services, views


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

    app = aiohttp.Application(web.Application)

    config = providers.Configuration()

    giphy_client = providers.Factory(
        giphy.GiphyClient,
        api_key=config.giphy.api_key,
        timeout=config.giphy.request_timeout,
    )

    search_service = providers.Factory(
        services.SearchService,
        giphy_client=giphy_client,
    )

    index_view = aiohttp.View(
        views.index,
        search_service=search_service,
    )


تأكد من تشغيل التطبيق أو تشغيله: 



adev runserver giphynavigator/application.py --livereload


وقم بتقديم طلب إلى API في المحطة: 



http http://localhost:8000/ query=="wow,it works" limit==5


سوف ترى: 



HTTP/1.1 200 OK
Content-Length: 850
Content-Type: application/json; charset=utf-8
Date: Wed, 29 Jul 2020 22:22:55 GMT
Server: Python/3.8 aiohttp/3.6.2

{
    "gifs": [
        {
            "url": "https://giphy.com/gifs/discoverychannel-nugget-gold-rush-rick-ness-KGGPIlnC4hr4u2s3pY"
        },
        {
            "url": "https://giphy.com/gifs/primevideoin-ll1hyBS2IrUPLE0E71"
        },
        {
            "url": "https://giphy.com/gifs/jackman-works-jackmanworks-l4pTgQoCrmXq8Txlu"
        },
        {
            "url": "https://giphy.com/gifs/cat-massage-at-work-l46CzMaOlJXAFuO3u"
        },
        {
            "url": "https://giphy.com/gifs/everwhatproductions-fun-christmas-3oxHQCI8tKXoeW4IBq"
        },
    ],
    "limit": 10,
    "query": "wow,it works"
}






البحث يعمل.



قليلا من إعادة بناء ديون



indexتحتوي طريقة العرض الخاصة بنا على قيمتين مشفرتين:



  • مصطلح البحث الافتراضي
  • حد لعدد النتائج


لنقم بإعادة بناء ديون صغيرة. سننقل هذه القيم إلى التكوين.



تحرير views.py: 



"""Views module."""

from aiohttp import web

from .services import SearchService


async def index(
        request: web.Request,
        search_service: SearchService,
        default_query: str,
        default_limit: int,
) -> web.Response:
    query = request.query.get('query', default_query)
    limit = int(request.query.get('limit', default_limit))

    gifs = await search_service.search(query, limit)

    return web.json_response(
        {
            'query': query,
            'limit': limit,
            'gifs': gifs,
        },
    )


الآن نحن بحاجة إلى تمرير هذه القيم عند الطلب. لنقم بتحديث الحاوية.



تحرير containers.py: 



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, services, views


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

    app = aiohttp.Application(web.Application)

    config = providers.Configuration()

    giphy_client = providers.Factory(
        giphy.GiphyClient,
        api_key=config.giphy.api_key,
        timeout=config.giphy.request_timeout,
    )

    search_service = providers.Factory(
        services.SearchService,
        giphy_client=giphy_client,
    )

    index_view = aiohttp.View(
        views.index,
        search_service=search_service,
        default_query=config.search.default_query,
        default_limit=config.search.default_limit,
    )


لنقم الآن بتحديث ملف التكوين.



تحرير config.yml: 



giphy:
  request_timeout: 10
search:
  default_query: "Dependency Injector"
  default_limit: 10


تم الانتهاء من إعادة البناء. لقد جعلنا تطبيقنا أكثر نظافة عن طريق نقل القيم المشفرة في التكوين.



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



إضافة الاختبارات



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



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



./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── giphy.py
│   ├── services.py
│   ├── tests.py
│   └── views.py
├── venv/
└── requirements.txt


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



"""Tests module."""

from unittest import mock

import pytest

from giphynavigator.application import create_app
from giphynavigator.giphy import GiphyClient


@pytest.fixture
def app():
    return create_app()


@pytest.fixture
def client(app, aiohttp_client, loop):
    return loop.run_until_complete(aiohttp_client(app))


async def test_index(client, app):
    giphy_client_mock = mock.AsyncMock(spec=GiphyClient)
    giphy_client_mock.search.return_value = {
        'data': [
            {'url': 'https://giphy.com/gif1.gif'},
            {'url': 'https://giphy.com/gif2.gif'},
        ],
    }

    with app.container.giphy_client.override(giphy_client_mock):
        response = await client.get(
            '/',
            params={
                'query': 'test',
                'limit': 10,
            },
        )

    assert response.status == 200
    data = await response.json()
    assert data == {
        'query': 'test',
        'limit': 10,
        'gifs': [
            {'url': 'https://giphy.com/gif1.gif'},
            {'url': 'https://giphy.com/gif2.gif'},
        ],
    }


async def test_index_no_data(client, app):
    giphy_client_mock = mock.AsyncMock(spec=GiphyClient)
    giphy_client_mock.search.return_value = {
        'data': [],
    }

    with app.container.giphy_client.override(giphy_client_mock):
        response = await client.get('/')

    assert response.status == 200
    data = await response.json()
    assert data['gifs'] == []


async def test_index_default_params(client, app):
    giphy_client_mock = mock.AsyncMock(spec=GiphyClient)
    giphy_client_mock.search.return_value = {
        'data': [],
    }

    with app.container.giphy_client.override(giphy_client_mock):
        response = await client.get('/')

    assert response.status == 200
    data = await response.json()
    assert data['query'] == app.container.config.search.default_query()
    assert data['limit'] == app.container.config.search.default_limit()


لنبدأ الآن في الاختبار والتحقق من التغطية: 



py.test giphynavigator/tests.py --cov=giphynavigator


سوف ترى: 



platform darwin -- Python 3.8.3, pytest-5.4.3, py-1.9.0, pluggy-0.13.1
plugins: cov-2.10.0, aiohttp-0.3.0, asyncio-0.14.0
collected 3 items

giphynavigator/tests.py ...                                     [100%]

---------- coverage: platform darwin, python 3.8.3-final-0 -----------
Name                            Stmts   Miss  Cover
---------------------------------------------------
giphynavigator/__init__.py          0      0   100%
giphynavigator/__main__.py          5      5     0%
giphynavigator/application.py      10      0   100%
giphynavigator/containers.py       10      0   100%
giphynavigator/giphy.py            16     11    31%
giphynavigator/services.py          9      1    89%
giphynavigator/tests.py            35      0   100%
giphynavigator/views.py             7      0   100%
---------------------------------------------------
TOTAL                              92     17    82%


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



يتم العمل. الآن دعونا نلخص.



خاتمة



لقد قمنا ببناء تطبيق aiohttpREST API باستخدام مبدأ حقن التبعية. استخدمنا حاقن التبعية كإطار لحقن التبعية.



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



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



"""Application containers module."""

from dependency_injector import containers, providers
from dependency_injector.ext import aiohttp
from aiohttp import web

from . import giphy, services, views


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

    app = aiohttp.Application(web.Application)

    config = providers.Configuration()

    giphy_client = providers.Factory(
        giphy.GiphyClient,
        api_key=config.giphy.api_key,
        timeout=config.giphy.request_timeout,
    )

    search_service = providers.Factory(
        services.SearchService,
        giphy_client=giphy_client,
    )

    index_view = aiohttp.View(
        views.index,
        search_service=search_service,
        default_query=config.search.default_query,
        default_limit=config.search.default_limit,
    )




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



ماذا بعد؟





More articles:


All Articles