Aiohttp + Dependency Injector - Tutorial zur Abhängigkeitsinjektion

Hallo,



ich bin der Schöpfer von Dependency Injector . Dies ist ein Abhängigkeitsinjektionsframework für Python.



Fortsetzung einer Reihe von Tutorials zur Verwendung des Abhängigkeitsinjektors zum Erstellen von Anwendungen.



In diesem Tutorial möchte ich zeigen, wie der Abhängigkeitsinjektor für die aiohttpAnwendungsentwicklung verwendet wird .



Das Handbuch besteht aus folgenden Teilen:

1. Was werden wir bauen?
2. Vorbereitung der Umwelt
3. Projektstruktur
4. Abhängigkeiten installieren
5. Minimale Anwendung
6. Giphy API-Client
7. Suchdienst
8. Suche verbinden
9. Ein bisschen Refactoring
10. Tests hinzufügen
11. Fazit

Das abgeschlossene Projekt finden Sie auf Github .



Um zu beginnen, müssen Sie haben:

• Python 3.5+
• Virtuelle Umgebung

Und es ist wünschenswert zu haben:

• Erste Entwicklungsfähigkeiten mit aiohttp
• Das Prinzip der Abhängigkeitsinjektion verstehen

Was werden wir bauen?

Wir werden eine REST-API-Anwendung erstellen, die auf Giphy nach lustigen Gifs sucht . Nennen wir es Giphy Navigator.



Wie funktioniert Giphy Navigator?

• Der Client sendet eine Anfrage, in der angegeben wird, wonach gesucht und wie viele Ergebnisse zurückgegeben werden sollen.
• Giphy Navigator gibt eine JSON-Antwort zurück.
• Die Antwort beinhaltet:
  
  
  • Suchanfrage
  • Anzahl der Ergebnisse
  • GIF-URL-Liste

Beispielantwort:

{
    "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"
        }
    ]
}

Bereiten Sie die Umgebung vor

Beginnen wir mit der Vorbereitung der Umgebung.



Zunächst müssen wir einen Projektordner und eine virtuelle Umgebung erstellen:

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

Jetzt aktivieren wir die virtuelle Umgebung:

. venv/bin/activate

Die Umgebung ist fertig, jetzt beginnen wir mit der Projektstruktur.

Projektstruktur

In diesem Abschnitt organisieren wir die Struktur des Projekts.



Lassen Sie uns die folgende Struktur im aktuellen Ordner erstellen. Lassen Sie vorerst alle Dateien leer.



Anfangsstruktur:

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

Abhängigkeiten installieren

Es ist Zeit, die Abhängigkeiten zu installieren. Wir werden Pakete wie dieses verwenden:

• dependency-injector - Abhängigkeitsinjektions-Framework
• aiohttp - Web-Framework
• aiohttp-devtools - Eine Hilfsbibliothek, die einen Server für die Live-Neustartentwicklung bereitstellt
• pyyaml - Bibliothek zum Parsen von YAML-Dateien, die zum Lesen der Konfiguration verwendet wird
• pytest-aiohttp- Hilfsbibliothek zum Testen von aiohttpAnwendungen
• pytest-cov - Hilfsbibliothek zur Messung der Codeabdeckung durch Tests

Fügen wir der Datei die folgenden Zeilen hinzu requirements.txt:

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

Und im Terminal ausführen:

pip install -r requirements.txt

Zusätzlich installieren httpie. Es ist ein Befehlszeilen-HTTP-Client. Wir werden

es verwenden, um die API manuell zu testen.



Lassen Sie uns im Terminal ausführen:

pip install httpie

Die Abhängigkeiten werden installiert. Jetzt erstellen wir eine minimale Anwendung.

Minimale Anwendung

In diesem Abschnitt erstellen wir eine minimale Anwendung. Es wird einen Endpunkt haben, der eine leere Antwort zurückgibt.



Lassen Sie uns bearbeiten 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,
        },
    )

Fügen wir nun einen Container für Abhängigkeiten hinzu (im Folgenden nur einen Container). Der Container enthält alle Komponenten der Anwendung. Fügen wir die ersten beiden Komponenten hinzu. Dies ist eine aiohttpAnwendung und Präsentation index.



Lassen Sie uns bearbeiten 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)

Jetzt müssen wir eine aiohttpAnwendungsfactory erstellen . Es wird normalerweise genannt

create_app(). Es wird ein Container erstellt. Der Container wird zum Erstellen der aiohttpAnwendung verwendet. Der letzte Schritt besteht darin, das Routing einzurichten. Wir weisen index_viewdem Container eine Ansicht zu, um Anforderungen an das Stammverzeichnis "/"unserer Anwendung zu verarbeiten.



Lassen Sie uns bearbeiten 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

Der Container ist das erste Objekt in der Anwendung. Es wird verwendet, um alle anderen Objekte abzurufen.

Jetzt können wir unsere Anwendung starten:



Führen Sie den Befehl im Terminal aus:

adev runserver giphynavigator/application.py --livereload

Die Ausgabe sollte folgendermaßen aussehen:

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

Wir verwenden httpie, um den Serverbetrieb zu überprüfen:

http http://127.0.0.1:8000/

Du wirst sehen:

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

Die minimale Anwendung ist fertig. Verbinden wir die Giphy-API.

Giphy API-Client

In diesem Abschnitt werden wir unsere Anwendung in die Giphy-API integrieren. Wir werden unseren eigenen API-Client auf der Client-Seite erstellen aiohttp.



Erstellen Sie eine leere Datei giphy.pyim Paket giphynavigator:

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

und fügen Sie die folgenden Zeilen hinzu:

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

Jetzt müssen wir den GiphyClient zum Container hinzufügen. GiphyClient hat zwei Abhängigkeiten, die beim Erstellen übergeben werden müssen: API-Schlüssel und Anforderungszeitlimit. Dazu müssen wir zwei neue Anbieter aus dem Modul verwenden dependency_injector.providers:

• Der Anbieter Factoryerstellt den GiphyClient.
• Der Anbieter Configurationsendet den API-Schlüssel und das Timeout an den GiphyClient.

Lassen Sie uns bearbeiten 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)

Wir haben die Konfigurationsparameter verwendet, bevor wir ihre Werte eingestellt haben. Dies ist das Prinzip, nach dem der Anbieter arbeitet Configuration.



Zuerst verwenden wir, dann setzen wir die Werte.

Fügen wir nun die Konfigurationsdatei hinzu.

Wir werden YAML verwenden.



Erstellen Sie eine leere Datei config.ymlim Stammverzeichnis des Projekts:

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

Und füllen Sie es mit den folgenden Zeilen aus:

giphy:
  request_timeout: 10

Wir werden eine Umgebungsvariable verwenden, um den API-Schlüssel zu übergeben GIPHY_API_KEY .



Jetzt müssen wir bearbeiten create_app(), um 2 Aktionen auszuführen, wenn die Anwendung gestartet wird:

• Konfiguration laden von config.yml
• Laden Sie den API-Schlüssel aus der Umgebungsvariablen GIPHY_API_KEY

Bearbeiten 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

Jetzt müssen wir einen API-Schlüssel erstellen und ihn auf eine Umgebungsvariable setzen.



Verwenden Sie jetzt diesen Schlüssel, um keine Zeit damit zu verschwenden:

export GIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0

Befolgen Sie dieses Tutorial , um Ihren eigenen Giphy-API-Schlüssel zu erstellen .

Die Erstellung und Konfiguration der Giphy API-Clients ist abgeschlossen. Fahren wir mit dem Suchdienst fort.

Suchdienst

Es ist Zeit, einen Suchdienst hinzuzufügen SearchService. Er wird:

• Suche
• Formatieren Sie die empfangene Antwort

SearchServicewird verwenden GiphyClient.



Erstellen Sie eine leere Datei services.pyim Paket giphynavigator:

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

und fügen Sie die folgenden Zeilen hinzu:

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

Beim Erstellen müssen SearchServiceSie übertragen GiphyClient. Wir werden dies angeben, wenn wir es SearchServicedem Container hinzufügen .



Lassen Sie uns bearbeiten 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)

Der Suchdienst ist jetzt SearchServiceabgeschlossen. Im nächsten Abschnitt verbinden wir es mit unserer Ansicht.

Suche verbinden

Wir sind jetzt bereit für die Suche. Lassen Sie uns SearchServicein indexSicht verwenden.



Bearbeiten 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,
        },
    )

Ändern wir nun den Container, um die Abhängigkeit beim Aufruf SearchServicean die Ansicht zu übergeben index.



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

Stellen Sie sicher, dass die Anwendung ausgeführt wird oder ausgeführt wird:

adev runserver giphynavigator/application.py --livereload

und stellen Sie eine Anfrage an die API im Terminal:

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

Du wirst sehen:

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

Die Suche funktioniert.

Ein bisschen Refactoring

Unsere Ansicht indexenthält zwei fest codierte Werte:

• Standard-Suchbegriff
• Begrenzen Sie die Anzahl der Ergebnisse

Lassen Sie uns ein wenig umgestalten. Wir werden diese Werte in die Konfiguration übertragen.



Bearbeiten 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,
        },
    )

Jetzt müssen diese Werte auf Abruf übergeben werden. Lassen Sie uns den Container aktualisieren.



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

Jetzt aktualisieren wir die Konfigurationsdatei.



Bearbeiten config.yml:

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

Das Refactoring ist abgeschlossen. Wir haben unsere Anwendung sauberer gemacht, indem wir fest codierte Werte in die Konfiguration verschoben haben.



Im nächsten Abschnitt werden wir einige Tests hinzufügen.

Tests hinzufügen

Es wäre schön, einige Tests hinzuzufügen. Machen wir das. Wir werden Pytest und Coverage verwenden .



Erstellen Sie eine leere Datei tests.pyim Paket giphynavigator:

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

und fügen Sie die folgenden Zeilen hinzu:

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

Beginnen wir nun mit dem Testen und überprüfen Sie die Abdeckung:

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

Du wirst sehen:

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%

Beachten Sie, wie wir giphy_client mit der Methode durch Mock ersetzen .override(). Auf diese Weise können Sie den Rückgabewert eines beliebigen Anbieters überschreiben.

Die Arbeit ist erledigt. Lassen Sie uns nun zusammenfassen.

Fazit

Wir haben eine aiohttpREST-API-Anwendung nach dem Prinzip der Abhängigkeitsinjektion erstellt. Wir haben Dependency Injector als Abhängigkeitsinjektions-Framework verwendet.



Der Vorteil, den Sie mit Dependency Injector erhalten, ist der Container.



Der Container beginnt sich auszuzahlen, wenn Sie die Struktur Ihrer Anwendung verstehen oder ändern müssen. Mit einem Container ist dies einfach, da sich alle Komponenten der Anwendung und ihre Abhängigkeiten an einem Ort befinden:

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

Ein Container als Karte Ihrer Anwendung. Sie wissen immer, was von was abhängt.

Was weiter?

• Erfahren Sie mehr über Dependency Injector auf GitHub
• Lesen Sie die Dokumentation unter Lesen Sie die Dokumente
• Haben Sie eine Frage oder finden Sie einen Fehler? Öffnen Sie eine Ausgabe auf Github