Aiohttp tutorial

This tutorial shows how to build anaiohttp REST API application following the dependencyinjection principle.

Start from the scratch or jump to the section:

You can find complete project on theGithub.

What are we going to build?

https://media.giphy.com/media/apvx5lPCPsjN6/source.gif

We will build a REST API application that searches for funny GIFs on theGiphy.Let’s call it Giphy Navigator.

How does Giphy Navigator work?

  • Client sends a request specifying the search query and the number of results.

  • Giphy Navigator returns a response in json format.

  • The response contains:

    • the search query

    • the limit number

    • the list of gif urls

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

The task is naive and that’s exactly what we need for the tutorial.

Prepare the environment

Let’s create the environment for the project.

First we need to create a project folder:

mkdirgiphynav-aiohttp-tutorialcdgiphynav-aiohttp-tutorial

Now let’s create and activate virtual environment:

python3-mvenvvenv.venv/bin/activate

Environment is ready and now we’re going to create the layout of the project.

Project layout

Create next structure in the current directory. All files should be empty. That’s ok for now.

Initial project layout:

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

Install the requirements

Now it’s time to install the project requirements. We will use next packages:

  • dependency-injector - the dependency injection framework

  • aiohttp - the web framework

  • pyyaml - the YAML files parsing library, used for the reading of the configuration files

  • pytest-aiohttp - the helper library for the testing of theaiohttp application

  • pytest-cov - the helper library for measuring the test coverage

Put next lines into therequirements.txt file:

dependency-injectoraiohttppyyamlpytest-aiohttppytest-cov

and run next in the terminal:

pipinstall-rrequirements.txt

Let’s also install thehttpie. It is a user-friendly command-line HTTP client for the API era.We will use it for the manual testing.

Run the command in the terminal:

pipinstallhttpie

The requirements are setup. Now we will build a minimal application.

Minimal application

In this section we will build a minimal application. It will have an endpoint thatwill answer our requests in json format. There will be no payload for now.

Edithandlers.py:

"""Handlers module."""fromaiohttpimportwebasyncdefindex(request:web.Request)->web.Response:query=request.query.get("query","Dependency Injector")limit=int(request.query.get("limit",10))gifs=[]returnweb.json_response({"query":query,"limit":limit,"gifs":gifs,},)

Now let’s create a container. Container will keep all of the application components and their dependencies.

Editcontainers.py:

"""Containers module."""fromdependency_injectorimportcontainersclassContainer(containers.DeclarativeContainer):...

Container is empty for now. We will add the providers in the following sections.

Finally we need to createaiohttp application factory. It will create and configure containerandweb.Application. It is traditionally calledcreate_app().We will assignindex handler to handle user requests to the root/ of our web application.

Put next into theapplication.py:

"""Application module."""fromaiohttpimportwebfrom.containersimportContainerfrom.importhandlersdefcreate_app()->web.Application:container=Container()app=web.Application()app.container=containerapp.add_routes([web.get("/",handlers.index),])returnappif__name__=="__main__":app=create_app()web.run_app(app)

Now we’re ready to run our application

Do next in the terminal:

python-mgiphynavigator.application

The output should be something like:

========Runningonhttp://0.0.0.0:8080========(PressCTRL+Ctoquit)

Let’s check that it works. Open another terminal session and usehttpie:

httphttp://0.0.0.0:8080/

You should see:

HTTP/1.1200OKContent-Length:844Content-Type:application/json; charset=utf-8Date:Wed, 29 Jul 2020 21:01:50 GMTServer:Python/3.10 aiohttp/3.6.2{"gifs":[],"limit":10,"query":"Dependency Injector"}

Minimal application is ready. Let’s connect our application with 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 usingaiohttp client.

Creategiphy.py module in thegiphynavigator package:

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

and put next into it:

"""Giphy client module."""fromaiohttpimportClientSession,ClientTimeoutclassGiphyClient:API_URL="https://api.giphy.com/v1"def__init__(self,api_key,timeout):self._api_key=api_keyself._timeout=ClientTimeout(timeout)asyncdefsearch(self,query,limit):"""Make search API call and return result."""url=f"{self.API_URL}/gifs/search"params={"q":query,"api_key":self._api_key,"limit":limit,}asyncwithClientSession(timeout=self._timeout)assession:asyncwithsession.get(url,params=params)asresponse:ifresponse.status!=200:response.raise_for_status()returnawaitresponse.json()

Now we need to addGiphyClient into the container. TheGiphyClient has two dependenciesthat have to be injected: the API key and the request timeout. We will need to use two moreproviders from thedependency_injector.providers module:

  • Factory provider. It will create aGiphyClient client.

  • Configuration provider. It will provide an API key and a request timeout for theGiphyClientclient. We will specify the location of the configuration file. The configuration provider will parsethe configuration file when we create a container instance.

Editcontainers.py:

"""Containers module."""fromdependency_injectorimportcontainers,providersfrom.importgiphyclassContainer(containers.DeclarativeContainer):config=providers.Configuration(yaml_files=["config.yml"])giphy_client=providers.Factory(giphy.GiphyClient,api_key=config.giphy.api_key,timeout=config.giphy.request_timeout,)

Now let’s add the configuration file. We will use YAML. Create an empty fileconfig.yml inthe root root of the project:

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

and put next into it:

giphy:request_timeout:10

We will use theGIPHY_API_KEY environment variable to provide the API key. Let’s editcreate_app() to fetch the key value from it.

Editapplication.py:

"""Application module."""fromaiohttpimportwebfrom.containersimportContainerfrom.importhandlersdefcreate_app()->web.Application:container=Container()container.config.giphy.api_key.from_env("GIPHY_API_KEY")app=web.Application()app.container=containerapp.add_routes([web.get("/",handlers.index),])returnappif__name__=="__main__":app=create_app()web.run_app(app)

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

As for now, don’t worry, just take this one:

exportGIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0

Note

To create your own Giphy API key follow thisguide.

The Giphy API client and the configuration setup is done. Let’s proceed to the search service.

Search service

Now it’s time to add theSearchService. It will:

  • Perform the search.

  • Format result data.

SearchService will useGiphyClient.

Createservices.py module in thegiphynavigator package:

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

and put next into it:

"""Services module."""from.giphyimportGiphyClientclassSearchService:def__init__(self,giphy_client:GiphyClient):self._giphy_client=giphy_clientasyncdefsearch(self,query,limit):"""Search for gifs and return formatted data."""ifnotquery:return[]result=awaitself._giphy_client.search(query,limit)return[{"url":gif["url"]}forgifinresult["data"]]

TheSearchService has a dependency on theGiphyClient. This dependency will beinjected when we addSearchService to the container.

Editcontainers.py:

"""Containers module."""fromdependency_injectorimportcontainers,providersfrom.importgiphy,servicesclassContainer(containers.DeclarativeContainer):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,)

The search service is ready. In next section we’re going to put it to work.

Make the search work

Now we are ready to put the search into work. Let’s injectSearchService intotheindex handler. We will useWiring feature.

Edithandlers.py:

"""Handlers module."""fromaiohttpimportwebfromdependency_injector.wiringimportProvide,injectfrom.servicesimportSearchServicefrom.containersimportContainer@injectasyncdefindex(request:web.Request,search_service:SearchService=Provide[Container.search_service],)->web.Response:query=request.query.get("query","Dependency Injector")limit=int(request.query.get("limit",10))gifs=awaitsearch_service.search(query,limit)returnweb.json_response({"query":query,"limit":limit,"gifs":gifs,},)

To make the injection work we need to wire the container with thehandlers module.Let’s configure the container to automatically make wiring with thehandlers module when wecreate a container instance.

Editcontainers.py:

"""Containers module."""fromdependency_injectorimportcontainers,providersfrom.importgiphy,servicesclassContainer(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,)

Make sure the app is running:

python-mgiphynavigator.application

and make a request to the API in the terminal:

httphttp://0.0.0.0:8080/query=="wow,it works"limit==5

You should see:

HTTP/1.1200OKContent-Length:492Content-Type:application/json; charset=utf-8Date:Fri, 09 Oct 2020 01:35:48 GMTServer:Python/3.10 aiohttp/3.6.2{"gifs":[{"url":"https://giphy.com/gifs/dollyparton-3xIVVMnZfG3KQ9v4Ye"},{"url":"https://giphy.com/gifs/tennistv-unbelievable-disbelief-cant-believe-UWWJnhHHbpGvZOapEh"},{"url":"https://giphy.com/gifs/discoverychannel-nugget-gold-rush-rick-ness-KGGPIlnC4hr4u2s3pY"},{"url":"https://giphy.com/gifs/soulpancake-wow-work-xUe4HVXTPi0wQ2OAJC"},{"url":"https://giphy.com/gifs/readingrainbow-teamwork-levar-burton-reading-rainbow-3o7qE1EaTWLQGDSabK"}],"limit":5,"query":"wow,it works"}
https://media.giphy.com/media/3oxHQCI8tKXoeW4IBq/source.gif

The search works!

Make some refactoring

Ourindex handler has two hardcoded config values:

  • Default search query

  • Default results limit

Let’s make some refactoring. We will move these values to the config.

Edithandlers.py:

"""Handlers module."""fromaiohttpimportwebfromdependency_injector.wiringimportProvide,injectfrom.servicesimportSearchServicefrom.containersimportContainer@injectasyncdefindex(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=awaitsearch_service.search(query,limit)returnweb.json_response({"query":query,"limit":limit,"gifs":gifs,},)

Let’s update the config.

Editconfig.yml:

giphy:request_timeout:10default:query:"DependencyInjector"limit:10

The refactoring is done. We’ve made it cleaner - hardcoded values are now moved to the config.

Tests

In this section we will add some tests.

Createtests.py module in thegiphynavigator package:

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

and put next into it:

"""Tests module."""fromunittestimportmockimportpytestfromgiphynavigator.applicationimportcreate_appfromgiphynavigator.giphyimportGiphyClient@pytest.fixturedefapp():app=create_app()yieldappapp.container.unwire()@pytest.fixturedefclient(app,aiohttp_client,loop):returnloop.run_until_complete(aiohttp_client(app))asyncdeftest_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"},],}withapp.container.giphy_client.override(giphy_client_mock):response=awaitclient.get("/",params={"query":"test","limit":10,},)assertresponse.status==200data=awaitresponse.json()assertdata=={"query":"test","limit":10,"gifs":[{"url":"https://giphy.com/gif1.gif"},{"url":"https://giphy.com/gif2.gif"},],}asyncdeftest_index_no_data(client,app):giphy_client_mock=mock.AsyncMock(spec=GiphyClient)giphy_client_mock.search.return_value={"data":[],}withapp.container.giphy_client.override(giphy_client_mock):response=awaitclient.get("/")assertresponse.status==200data=awaitresponse.json()assertdata["gifs"]==[]asyncdeftest_index_default_params(client,app):giphy_client_mock=mock.AsyncMock(spec=GiphyClient)giphy_client_mock.search.return_value={"data":[],}withapp.container.giphy_client.override(giphy_client_mock):response=awaitclient.get("/")assertresponse.status==200data=awaitresponse.json()assertdata["query"]==app.container.config.default.query()assertdata["limit"]==app.container.config.default.limit()

Now let’s run it and check the coverage:

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

You should see:

platformdarwin--Python3.10.0,pytest-6.2.5,py-1.10.0,pluggy-1.0.0plugins:asyncio-0.16.0,anyio-3.3.4,aiohttp-0.3.0,cov-3.0.0collected3itemsgiphynavigator/tests.py...[100%]----------coverage:platformdarwin,python3.10.0-final-0----------NameStmtsMissCover---------------------------------------------------giphynavigator/__init__.py00100%giphynavigator/application.py13285%giphynavigator/containers.py70100%giphynavigator/giphy.py14936%giphynavigator/handlers.py100100%giphynavigator/services.py9189%giphynavigator/tests.py370100%---------------------------------------------------TOTAL901287%

Note

Take a look at the highlights in thetests.py.

It emphasizes the overriding of theGiphyClient. The real API call are mocked.

Conclusion

In this tutorial we’ve built anaiohttp REST API application following the dependencyinjection principle.We’ve used theDependencyInjector as a dependency injection framework.

Containers andProviders helped to specify how to assemble search service andgiphy client.

Configuration provider helped to deal with reading YAML file and environment variable.

We usedWiring feature to inject the dependencies into theindex() handler.Provider overriding feature helped in testing.

We kept all the dependencies injected explicitly. This will help when you need to add orchange something in future.

You can find complete project on theGithub.

What’s next?

Sponsor the project on GitHub: