Quick guide to Asyncio in Python

The purpose of this quick guide to asyncio in Python is to demonstrate the benefits of asynchronous operation, how to use Python asyncio and at the same time how to test asyncio.

Concurrent programming

In asynchronous or concurrent programming, your code is multitasking to finish some tasks that requires waiting, but the operation is done in a single threaded and single process scheme. This is in contrast to parallelism that is utilizing multiprocessing.

Asynchronous code is common in many programming languages like Javascript, Golang and Java. Although the Asyncio library was only introduced in Python version 3.4, using the async/await syntax.

Analogy in a concurrent code

You arrived at the mall with your girlfriend on the same car. To finish your shopping much quicker, you decided to buy your gaming console while she is in the fitting room. And after both of you are done shopping, you end up meeting in Starbucks to grab a drink and finally go back to the car.

Example use of Asyncio

from typing import List, Dict
import asyncio
import time

class Content:

  def __init__(self):
    pass

  async def get_posts(self) -> List[Dict[any, any]]:
    await asyncio.sleep(1) # Fake data retrieval
    return [{"id": 1, "title": "Lorem Ipsum"}]

  async def get_albums(self) -> List[Dict[any, any]]:
    await asyncio.sleep(1) # Fake data retrieval
    return [{"id": 1, "title": "Mona Lisa"}]

  async def load_contents(self) -> None:
    start_time = time.perf_counter()
    await asyncio.gather(
      self.get_posts(),
      self.get_albums(),
    )
    exec_time = (time.perf_counter() - start_time)
    print(f"Execution time: {exec_time:0.2f} seconds.")

  async def load_contents_using_tasks(self) -> None:
    start_time = time.perf_counter()
    content_types = ["posts", "albums"]
    tasks: List[asyncio.Task] = []
    for content in content_types:
      tasks.append(getattr(self, f"get_{content}")())
    await asyncio.gather(*tasks)
    exec_time = (time.perf_counter() - start_time)
    print(f"Execution time: {exec_time:0.2f} seconds.")

  async def load_contents_synchronous(self) -> None:
    start_time = time.perf_counter()
    await self.get_posts()
    await self.get_albums()
    exec_time = (time.perf_counter() - start_time)
    print(f"Execution time: {exec_time:0.2f} seconds.")

async def app() -> None:
  content = Content()
  await content.load_contents()
  await content.load_contents_using_tasks()
  await content.load_contents_synchronous()

if __name__ == "__main__":
  asyncio.run(app())

Let’s explain the key points in this example. To imitate a data retrieval process you will see inside the class methods self.get_posts() and self.get_albums() the asyncio.sleep(1). What it will do is put a delay process for about one second on each method.

Async and Await syntax

To declare Coroutines and consume awaitable task objects, the Class methods are defined using async def syntax, except for __init__ method. Consequently a call to an async operation is then prepended with await.

Running an async application

async def app() -> None:
  content = Content()
  await content.load_contents()
  await content.load_contents_using_tasks()
  await content.load_contents_synchronous()

if __name__ == "__main__":
  asyncio.run(app())

The entire application will be invoked using asyncio.run(app()) that calls the top-level async def app() function.

In content.load_contents(), I have manually added the calls to self.get_posts() and self.get_albums() as a parameter to await asyncio.gather(...). The process here will run in asynchronous operation, thus the execution time will shorten to one second.

Similar to content.load_contents(), however this time the content.load_contents_using_tasks() is creating tasks dynamically using asyncio.create_task.

Unlike the first two methods, content.load_contents_synchronous() is waiting for self.get_posts() and self.get_albums() to finish synchronously. The tasks does not overlap and has to wait for the first task object to finish before proceeding to the next one. As a result the execution time will double to two seconds.

The result of async

In asynchronous operation, the runtime finished in just 1.00 second, which is half the time of the normal process. Hence, if asyncio library is used properly, this will absolutely help in improving the performance of an application.

python asyncapp.py
Execution time: 1.00 seconds.
Execution time: 1.00 seconds.
Execution time: 2.00 seconds.

Testing an async code

What good is an application if we are not able to test it. Normally this would be more sophisticated test case and we will have to mock the data, but in here I just want to demonstrate on how we can execute the Coroutines in a test.

Thanks to a solution that I found online, we can mock the call to asyncio.sleep(1) using our class AsyncMock since we do not want the wait during tests right.

from unittest import TestCase
from unittest.mock import patch, MagicMock
from asyncapp import Content
import asyncio

class AsyncMock(MagicMock):
  async def __call__(self, *args, **kwargs):
    return super(AsyncMock, self).__call__(*args, **kwargs)

@patch("asyncapp.asyncio.sleep", new_callable=AsyncMock)
class TestContent(TestCase):
  @classmethod
  def setUpClass(cls):
    cls.content = Content()
    cls.event_loop = asyncio.get_event_loop()

  @classmethod
  def tearDownClass(cls):
    cls.event_loop.close()

  def test_get_posts(self, async_mock):
    data = self.event_loop.run_until_complete(self.content.get_posts())
    async_mock.assert_called_with(1)
    self.assertTrue(data)

  def test_get_albums(self, async_mock):
    data = self.event_loop.run_until_complete(self.content.get_albums())
    async_mock.assert_called_with(1)
    self.assertTrue(data)

Our patch will apply in class level that will allow the use of AsyncMock in all the test methods. Using the key parameter new_callable, we are able to replace the actual sleep function with a Mock class.

@patch("asyncapp.asyncio.sleep", new_callable=AsyncMock)
class TestContent(TestCase):

Inside setUpClass we created an event loop with asyncio.get_event_loop(). As a result, it is now possible to test the async methods in conjunction with the self.event_loop.run_until_complete function.

data = self.event_loop.run_until_complete(self.content.get_posts())
...
data = self.event_loop.run_until_complete(self.content.get_albums())