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."""
fromdependency_injectorimport containers, providers
from.import giphy, services
classContainer(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."""
fromaiohttpimport web
fromdependency_injector.wiringimport inject, Provide
from.servicesimport SearchService
from.containersimport Container
@inject
async defindex(
 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."""
fromaiohttpimport web
from.containersimport Container
from.import handlers
defcreate_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."""
fromunittestimport mock
importpytest
importpytest_asyncio
fromgiphynavigator.applicationimport create_app
fromgiphynavigator.giphyimport GiphyClient
pytestmark = pytest.mark.asyncio
@pytest.fixture
defapp():
 app = create_app()
 yield app
 app.container.unwire()
@pytest_asyncio.fixture
async defclient(app, aiohttp_client):
 return await aiohttp_client(app)
async deftest_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 deftest_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 deftest_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:

[フレーム]