Aiohttp + DependencyInjector-依存関係インジェクションチュートリアル

こんにちは、



私はDependencyInjectorの作成者です。これは、Pythonの依存関係注入フレームワークです。



DependencyInjectorを使用してアプリケーションを構築するための一連のチュートリアルを継続します。



このチュートリアルでは、aiohttpアプリケーション開発にDependencyInjectorを使用する方法を紹介します。



マニュアルは次の部分で構成されています。

1. 何を構築しますか？
2. 環境の準備
3. プロジェクト構造
4. 依存関係のインストール
5. 最小限のアプリケーション
6. GiphyAPIクライアント
7. 検索サービス
8. 検索を接続する
9. 少しリファクタリング
10. テストの追加
11. 結論

完成したプロジェクトはGithubにあります。



開始するには、次のものが必要です。

• Python 3.5+
• 仮想環境

そして、次のことが望ましいです。

• aiohttpによる初期開発スキル
• 依存関係の注入の原理を理解する

何を構築しますか？

Giphy で面白いgifを検索するRESTAPIアプリケーションを構築します。それをGiphyNavigatorと呼びましょう。



Giphy Navigatorはどのように機能しますか？

• クライアントは、何を検索し、いくつの結果を返すかを示す要求を送信します。
• GiphyNavigatorはjson応答を返します。
• 答えは次のとおりです。
  
  
  • 検索クエリ
  • 結果の数
  • GIFURLリスト

応答例：

{
    "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 -Webフレームワーク
• aiohttp-devtools -ライブリブート開発用のサーバーを提供するヘルパーライブラリ
• pyyaml -構成の読み取りに使用されるYAMLファイルを解析するためのライブラリ
• 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,
        },
    )

次に、依存関係のコンテナを追加しましょう（さらに単なるコンテナ）コンテナには、アプリケーションのすべてのコンポーネントが含まれます。最初の2つのコンポーネントを追加しましょう。これは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"
}

最小限のアプリケーションの準備ができています。GiphyAPIを接続しましょう。

GiphyAPIクライアント

このセクションでは、アプリケーションをGiphyAPIと統合します。クライアント側を使用して独自の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には、作成時に渡す必要のある2つの依存関係があります。APIキーと要求タイムアウトです。これを行うには、モジュールから2つの新しいプロバイダーを使用する必要があります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()、アプリケーションの起動時に2つのアクションを実行するように編集する必要があります。

• から構成をロード 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

このチュートリアルに従って、独自のGiphyAPIキーを作成します。

GiphyAPIクライアントの作成と構成のセットアップが完了しました。検索サービスに移りましょう。

検索サービス

検索サービスを追加する時が来ました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は、ハードコードされた2つの値が含まれています。

• デフォルトの検索語
• 結果数の制限

少しリファクタリングしてみましょう。これらの値を構成に転送します。



編集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とcoverageを使用します。パッケージに



空のファイル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 メソッドを使用してモックに置き換える方法に注目してください.override()。このようにして、任意のプロバイダーの戻り値をオーバーライドできます。

作業は完了です。それでは要約しましょう。

結論

aiohttp依存関係注入の原則を使用してRESTAPIアプリケーションを 構築しました。依存関係インジェクターフレームワークとして依存関係インジェクターを使用しました。



Dependency Injectorで得られる利点は、コンテナーです。



アプリケーションの構造を理解または変更する必要がある場合、コンテナは成果を上げ始めます。コンテナを使用すると、アプリケーションのすべてのコンポーネントとその依存関係が1か所にあるため、簡単です。

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

アプリケーションのマップとしてのコンテナー。あなたは常に何が何に依存するかを知っています。

次は何ですか？

• GitHubのDependencyInjectorの詳細
• Read theDocsのドキュメントを確認してください
• 質問があるか、バグを見つけますか？Githubで問題を開く