Building Powerful Applications with Python Asyncio

This article is part of the Python Libraries Series. Find more below!

Asynchronous programming is a paradigm that allows for more efficient and scalable applications. Asyncio, a built-in library in Python, can be used to build such applications.

Let’s discover what is asynchronous programming, how it can be used to build powerful applications, and how to use asyncio.

Asynchronous Programming

Asynchronous programming is a way of writing code that allows a program to perform multiple tasks simultaneously without blocking the execution of other code. This type of programming is particularly useful when you’re working with I/O-bound tasks, such as reading from a database, sending network requests, or waiting for user input.

In traditional synchronous programming, when a task is started, the program waits for it to complete before moving on to the next task. This can be time-consuming, especially when working with I/O-bound tasks that can take a long time to complete.

With asynchronous programming, you can start a task and then move on to another task while the first task is running. When the first task completes, it notifies the program, and the program can then process the result and move on to the next task.

As an example, let’s consider a program that needs to download data from a remote server and then process that data. In synchronous programming, the program would download the data and then wait for the download to complete before processing the data. This can take a long time, especially if the download is slow.

In asynchronous programming, the program can start the download and then move on to processing other data while the download is in progress. When the download completes, the program can process the downloaded data.

Now, let me introduce some Python asyncio concepts.

Python Asyncio Concepts

Coroutines: Coroutines are functions that can be paused and resumed. They are used to write asynchronous code. In asyncio, coroutines are defined using the async def syntax, and they are called with the await keyword.

Event Loop: The event loop is the core of asyncio. It is responsible for scheduling and running coroutines, handling I/O operations, and managing callbacks. The event loop runs in a single thread, so it is lightweight and efficient.

Futures: Futures are objects that represent the result of an asynchronous operation that hasn’t been completed yet. They can be used to wait for the completion of an asynchronous task or to run a callback function when the operation completes. In asyncio, futures are represented by the asyncio.Future class. You can create a future by calling loop.create_future().

Tasks: Tasks are a higher-level abstraction built on top of coroutines and futures. They represent the execution of a coroutine in the event loop. Tasks can be created using the asyncio.create_task() function, which schedules the coroutine to run in the event loop and returns a Task object. Tasks can be awaited just like coroutines, and they also have methods to check their status, cancel them, or add callbacks to be executed when they are completed.

Getting Started with Asyncio

Let’s start with a basic example to understand the asyncio syntax. First, we need to import asyncio:

import asyncio

Then, let’s define a coroutine function using the async def syntax:

async def my_coroutine():
print("Starting coroutine")
await asyncio.sleep(1)
print("Coroutine finished")

Now, we can create an event loop. The event loop is responsible for managing the execution of coroutines and other asynchronous operations.

loop = asyncio.get_event_loop()

We can now schedule the coroutine to run on the event loop using the asyncio.create_task() method. This method creates a Task object that represents the execution of the coroutine.

task = asyncio.create_task(my_coroutine())

To run the event loop, we can use loop.run_until_complete(task). And finally, we can close the event loop using loop.close()

A more complex example

The example above was just here to make you understand the asyncio syntax. Now let’s move on to a more complex example. The idea of the example is to retrieve data from different data sources and process them, asynchronously obviously. Let’s start by importing the libraries we’ll need.

import asyncio
import random

We define now a coroutine function called fetch_data() that takes a URL as an argument. This coroutine simulates fetching data from a URL by sleeping for a random amount of time between 1 and 3 seconds using the asyncio.sleep() function. We print a message before and after the sleep to show when the coroutine starts and finishes.

async def fetch_data(url):
  print(f"Fetching data from {url}")
  await asyncio.sleep(random.randint(1, 3)) # Simulate a network delay
  print(f"Received data from {url}")
  return f"Data from {url}"

Then we define another coroutine function called process_data() that takes data as an argument. This coroutine simulates processing the data, also by sleeping for a random amount of time between 1 and 3 seconds.

async def process_data(data):
  print(f"Processing data {data}")
  await asyncio.sleep(random.randint(1, 3)) # Simulate a CPU-bound task
  print(f"Processed data {data}")
  return f"Processed data {data}"

Now, we define the main() coroutine function that creates a list of URLs to fetch data from, creates a task for each URL using the asyncio.create_task() function, and runs the tasks concurrently using the asyncio.gather() function. We print the results of the asyncio.gather() function to show the processed data.

async def main():
  urls = ["https://example.com", "https://google.com", "https://github.com"]
  tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
  fetched_data = await asyncio.gather(*tasks)
  processed_data = await asyncio.gather(*(process_data(data) for data in fetched_data))
  print(processed_data)

The last step is to run our program using asyncio.run(main())

The output should look like the following:

Fetching data from https://example.com
Fetching data from https://google.com
Fetching data from https://github.com
Received data from https://example.com
Received data from https://google.com
Received data from https://github.com
Processing data Data from https://example.com
Processing data Data from https://google.com
Processing data Data from https://github.com
Processed data Processed data Data from https://example.com
Processed data Processed data Data from https://google.com
Processed data Processed data Data from https://github.com
['Processed data Data from https://example.com', 'Processed data Data from https://google.com', 'Processed data Data from https://github.com']

Debugging Asynchronous Code

Debugging asynchronous code in Python can be a bit tricky due to the nature of the event loop and the way the code is executed. That’s why there are some guidelines to follow when you want to debug this type of code.

Using the asyncio debugger: The asyncio module provides a built-in debugger that can be used to debug your asynchronous code. You can enable the debugger by adding the following lines to your code:

import asyncio
asyncio.get_event_loop().set_debug(True)

This will enable debug output from the event loop, which can help you identify issues with your code.

Using print statements (or any third-party logging librairies): Sometimes the easiest way to debug your code is to add print statements to your code. Since asyncio code is executed asynchronously, adding print statements can help you understand the order in which your code is executed and identify any issues.

import asyncio

async def my_coroutine():
  print("Starting my_coroutine")
  await asyncio.sleep(1)
  print("Finished my_coroutine")

async def main():
  print("Starting main")
  await my_coroutine()
  print("Finished main")

asyncio.run(main())

Using a third-party debugger: Another option is to use a third-party debugger that is designed specifically for asyncio code. Two popular options are aiohttp-devtools and asyncio_debugger.

import aiohttp_devtools
aiohttp_devtools.setup()

import asyncio_debugger
asyncio_debugger.init()

Use breakpoints: If you are using an IDE like PyCharm or VSCode, you can set breakpoints in your asyncio code and use the debugger to step through your code.

import asyncio

async def my_coroutine():
  breakpoint()
  await asyncio.sleep(1)
  print("Finished my_coroutine")

async def main():
  breakpoint()
  await my_coroutine()
  print("Finished main")

asyncio.run(main())

Asyncio best practices

First, it’s important to design your application with concurrency in mind from the start, breaking it down into small, self-contained tasks that can be run concurrently. This requires careful planning of the flow of data and communication between tasks, as well as an understanding of the event loop and how it manages tasks.

Then, you should avoid blocking operations within your asynchronous code. This means avoiding I/O operations that might block the event loop, and instead using asyncio-compatible libraries or asynchronous APIs wherever possible. If you need to use a synchronous library or function, consider wrapping it in a separate thread or process to avoid blocking the event loop, else there’s no purpose in writing asynchronous code.

Thirdly, pay close attention to error handling and propagation. Exceptions raised in asynchronous code can be tricky to handle correctly, especially when dealing with multiple concurrent tasks. You should use asyncio’s built-in exception handling mechanisms to catch and propagate exceptions between tasks and coroutines, and to ensure that any errors are properly logged or reported to the user.

Lastly, it’s important to test your code thoroughly, including edge cases and error conditions. Asynchronous code can be more complex to test than synchronous code, and you should make use of asyncio-specific testing frameworks and tools to ensure that your code is working correctly.

Use Cases

Asynchronous code can be used to build powerful applications as it allows your program to run several tasks at the same time, but what could this be useful for?

Web scraping: Python asyncio can be used for web scraping as it allows the scraping of multiple websites simultaneously. This can be achieved by using the aiohttp library, which is an asynchronous HTTP client/server library.

Network programming: asyncio can be used for building high-performance networking applications. It provides a way to handle thousands of simultaneous connections in a single thread, which is useful for building real-time chat applications, multiplayer games, and more.

Distributed computing: asyncio provides a way to write distributed systems as it uses asynchronous code. This can be useful for building distributed data processing systems or building distributed web applications.

Asynchronous task scheduling: Python asyncio can be used for running background tasks, such as sending emails, processing files, etc…

Web development: the asyncio library provides support for asynchronous web frameworks such as aiohttp, which can be used to build web applications.

Final Note

Asynchronous programming can be very powerful depending on the application you want to build.

I hope to have provided you with a good knowledge base to start with asynchronous programming with asyncio!

If you want to discover other Python libraries, click below!

If you liked the story, don’t forget to clap and maybe follow me if you want to explore more of my content :)

You can also subscribe to me via email to be notified every time I publish a new story, just click here!

If you’re not subscribed to medium yet and wish to support me or get access to all my stories, you can use my link: