Aiohttp + Dependency Injector - dependency injection tutorial

Hi,



I am the creator of Dependency Injector . This is a dependency injection framework for Python.



Continuing a series of tutorials on using the Dependency Injector to build applications.



In this tutorial I want to show you how to use the Dependency Injector for aiohttpapplication development .



The manual consists of the following parts:

1. What are we going to build?
2. Preparing the environment
3. Project structure
4. Installing dependencies
5. Minimal application
6. Giphy API client
7. Search service
8. Connect search
9. A bit of refactoring
10. Adding tests
11. Conclusion

The completed project can be found on Github .



To start you must have:

• Python 3.5+
• Virtual environment

And it is desirable to have:

• Initial development skills with aiohttp
• Understanding the principle of dependency injection

What are we going to build?

We will be building a REST API application that searches for funny gifs on Giphy . Let's call it Giphy Navigator.



How does Giphy Navigator work?

• The client sends a request indicating what to look for and how many results to return.
• Giphy Navigator returns json response.
• The answer includes:
  
  
  • search query
  • number of results
  • GIF url list

Sample response:

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

Prepare the environment

Let's start by preparing the environment.



First of all, we need to create a project folder and a virtual environment:

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

Now let's activate the virtual environment:

. venv/bin/activate

The environment is ready, now let's start with the project structure.

Project structure

In this section, we will organize the structure of the project.



Let's create the following structure in the current folder. Leave all files empty for now.



Initial structure:

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

Installing dependencies

It's time to install the dependencies. We will use packages like this:

• dependency-injector - dependency injection framework
• aiohttp - web framework
• aiohttp-devtools - a helper library that provides a server for live reboot development
• pyyaml - library for parsing YAML files, used to read the config
• pytest-aiohttp- helper library for testing aiohttpapplications
• pytest-cov - helper library for measuring code coverage by tests

Let's add the following lines to the file requirements.txt:

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

And execute in the terminal:

pip install -r requirements.txt

Install additionally httpie. It is a command line HTTP client. We will

use it to manually test the API.



Let's execute in the terminal:

pip install httpie

The dependencies are installed. Now let's build a minimal application.

Minimal application

In this section, we will build a minimal application. It will have an endpoint that will return an empty response.



Let's edit 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,
        },
    )

Now let's add a container for dependencies (hereinafter just a container). The container will contain all the components of the application. Let's add the first two components. This is an aiohttpapplication and presentation index.



Let's edit 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)

Now we need to create an aiohttpapplication factory . It is usually called

create_app(). It will create a container. The container will be used to create the aiohttpapplication. The final step is to set up routing - we will assign a view index_viewfrom the container to handle requests to the root of "/"our application.



Let's edit 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

The container is the first object in the application. It is used to get all other objects.

We are now ready to launch our application:



Run the command in the terminal:

adev runserver giphynavigator/application.py --livereload

The output should look like this:

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

We use httpieto check the server operation:

http http://127.0.0.1:8000/

You will see:

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

The minimal application is ready. Let's connect the Giphy API.

Giphy API client

In this section, we will integrate our application with the Giphy API. We will create our own API client using the client side aiohttp.



Create an empty file giphy.pyin the package giphynavigator:

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

and add the following lines to it:

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

Now we need to add the GiphyClient to the container. GiphyClient has two dependencies that need to be passed when it is created: API key and request timeout. To do this, we will need to use two new providers from the module dependency_injector.providers:

• The provider Factorywill create the GiphyClient.
• The provider Configurationwill send the API key and timeout to the GiphyClient.

Let's edit 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)

We used the configuration parameters before setting their values. This is the principle by which the provider works Configuration.



First we use, then we set the values.

Now let's add the configuration file.

We will use YAML.



Create an empty file config.ymlat the root of the project:

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

And fill it in with the following lines:

giphy:
  request_timeout: 10

We'll use an environment variable to pass the API key GIPHY_API_KEY .



Now we need to edit create_app()to do 2 actions when the application starts:

• Load configuration from config.yml
• Load API key from environment variable GIPHY_API_KEY

Edit 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

Now we need to create an API key and set it to an environment variable.



In order not to waste time on this, now use this key:

export GIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0

Follow this tutorial to create your own Giphy API key .

Creation of the Giphy API client and installation of the configuration is complete. Let's move on to the search service.

Search service

It's time to add a search service SearchService. He will:

• Search
• Format received response

SearchServicewill use GiphyClient.



Create an empty file services.pyin the package giphynavigator:

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

and add the following lines to it:

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

When creating, SearchServiceyou need to transfer GiphyClient. We will indicate this when we add it SearchServiceto the container.



Let's edit 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)

The search service is now SearchServicecomplete. In the next section, we'll connect it to our view.

Connect search

We are now ready for the search to work. Let's use SearchServicein indexview.



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

Now let's change the container to pass the dependency SearchServiceto the view indexwhen it is called.



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

Make sure the application is running or run:

adev runserver giphynavigator/application.py --livereload

and make a request to the API in the terminal:

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

You will see:

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

The search works.

A bit of refactoring

Our view indexcontains two hardcoded values:

• Default search term
• Limit for the number of results

Let's do a little refactoring. We will transfer these values ​​into the configuration.



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

Now we need these values ​​to be passed on call. Let's update the container.



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

Now let's update the config file.



Edit config.yml:

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

The refactoring is complete. We've made our application cleaner by moving hardcoded values ​​into the configuration.



In the next section, we'll add some tests.

Adding tests

It would be nice to add some tests. Let's do it. We'll be using pytest and coverage .



Create an empty file tests.pyin the package giphynavigator:

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

and add the following lines to it:

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

Now let's start testing and check coverage:

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

You will see:

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%

Notice how we replace giphy_client with mock using the method .override(). In this way, you can override the return value of any provider.

The work is done. Now let's summarize.

Conclusion

We have built a aiohttpREST API application using the dependency injection principle. We used Dependency Injector as a dependency injection framework.



The advantage you get with Dependency Injector is container.



The container starts to pay off when you need to understand or change the structure of your application. With a container, it's easy because all the components of the application and their dependencies are in one place:

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

A container as a map of your application. You always know what depends on what.

What's next?

• Learn more about Dependency Injector on GitHub
• Check out the documentation at Read the Docs
• Have a question or find a bug? Open an issue on Github