Aiohttp example

This example shows how to use Dependency Injector with Aiohttp.

The example application is a REST API that searches for funny GIFs on the Giphy.

The source code is available on the Github.

Aiohttp tutorial demonstrates how to build this application step-by-step.

Application structure

Application has next structure:

./
├── giphynavigator/
│   ├── __init__.py
│   ├── application.py
│   ├── containers.py
│   ├── giphy.py
│   ├── handlers.py
│   ├── services.py
│   └── tests.py
├── config.yml
└── requirements.txt

Container

Declarative container is defined in giphynavigator/containers.py:

"""Containers module."""

from dependency_injector import containers, providers

from . import giphy, services


class Container(containers.DeclarativeContainer):

    wiring_config = containers.WiringConfiguration(modules=[".handlers"])

    config = providers.Configuration(yaml_files=["config.yml"])

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

Handlers

Handler has dependencies on search service and some config options. The dependencies are injected using Wiring feature.

Listing of giphynavigator/handlers.py:

"""Handlers module."""

from aiohttp import web
from dependency_injector.wiring import inject, Provide

from .services import SearchService
from .containers import Container


@inject
async def index(
        request: web.Request,
        search_service: SearchService = Provide[Container.search_service],
        default_query: str = Provide[Container.config.default.query],
        default_limit: int = Provide[Container.config.default.limit.as_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,
        },
    )

Application factory

Application factory creates container, wires it with the handlers module, creates Aiohttp app and setup routes.

Listing of giphynavigator/application.py:

"""Application module."""

from aiohttp import web

from .containers import Container
from . import handlers


def create_app() -> web.Application:
    container = Container()
    container.config.giphy.api_key.from_env("GIPHY_API_KEY")

    app = web.Application()
    app.container = container
    app.add_routes([
        web.get("/", handlers.index),
    ])
    return app


if __name__ == "__main__":
    app = create_app()
    web.run_app(app)

Tests

Tests use Provider overriding feature to replace giphy client with a mock giphynavigator/tests.py:

"""Tests module."""

from unittest import mock

import pytest

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


@pytest.fixture
def app():
    app = create_app()
    yield app
    app.container.unwire()


@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.default.query()
    assert data["limit"] == app.container.config.default.limit()

Sources

Explore the sources on the Github.

Sponsor the project on GitHub: