Use Python’s asyncio for Async Development and Testing

I began new work recently that exclusively uses Python code. As a result of many factors, the performance of either the API or scripts that I was managing was becoming more and more necessary. By default, the project didn’t have any async implementation, anywhere. With some help and practice I was able to get a hang of maintaining any previous code to perform faster and write newer, more efficient features, too. Here’s what I learned!

As a prerequisite, you may need to install `asyncio` and `pytest` into your project.

Managing Asyncio

A rule of thumb that is important to note here and with any async functionality is that it has to be asynchronously all the way up the call stack. In Python scripts, there’s usually a main() function responsible for calling everything else. This means that main() needs to be defined as async. Additionally, you will need to call it correctly:

import asyncio

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

Running Many Tasks in Parallel

An outstanding reason for using asyncio is that it can run tasks in parallel. This is naturally more efficient. What I’m working on requires for things to be processed in loops and oftentimes these are async, too. By gathering all of them at once before they’re awaited, code run time can be made significantly faster. Here’s an example!

async def get_items(keys: list[str]):
  tasks = []
  tasks = [self.get_item(key) for key in keys]
  items = await.asyncio.gather(*tasks)

  return items

Here, we are grouping a list of tasks together, running them asynchronously, and returning the result as one neat object called items, which would be a list of whatever is returned from get_item().

One way to validate that this is working correctly is by writing a print statement somewhere within get_item(). When running this code, if the loop is expected to run 5 times, then the print statement should pop up 5 times instantly (each of these is being run in parallel)  in the console as opposed to iteratively / one by one / as each one is processed.

Testing Async Features

Assuming pytest (but similar will apply), you will need to run each test where async functionality is used differently than before. However, it is pretty to-the-point as long as asyncio is installed already. Of course, your test function will need to be defined with async – however, you will also need to add an annotation a line above the definition. Here’s an example:

@pytest.mark.anyio
async def test_get_items():
  # Arrange / Act / Assert …

Closing Thoughts

I hope this is helpful to someone! If you have any questions, leave them below. I tried to keep this blog post short and concise. With some of the examples above I know you’ll be setup for success.