Basics of Python asyncio: Mastering Asynchronous Programming from Scratch

In modern software development, we often need to handle a large number of IO operations, such as network requests, file reading and writing, and database queries. Traditional synchronous programming methods can block the program during these operations, leading to poor performance. This is where asynchronous programming becomes particularly important. Python’s asyncio library is a powerful tool designed to address this issue.

What is asyncio?

asyncio is a standard library introduced in Python 3.4 that provides the infrastructure for writing asynchronous programs. You can think of asyncio as an efficient “task scheduler” that can intelligently switch to other tasks while the program waits for IO operations to complete, thereby maximizing the execution efficiency of the program.

Unlike traditional multithreaded programming, asyncio adopts a single-threaded concurrency model. This may sound contradictory, but in reality, this design avoids common issues such as race conditions and locks found in multithreaded programming while still achieving excellent performance.

What is a Coroutine?

To understand asyncio, we first need to grasp the concept of coroutines. A coroutine is a special type of function that can pause its execution and resume at an appropriate time.

Imagine you are cooking in the kitchen, needing to cook rice, stir-fry vegetables, and make soup. The traditional synchronous way is like strictly following the order: first wash the rice and put it in the rice cooker, then stand there waiting for 30 minutes until the rice is done, before starting to stir-fry the vegetables, and after that, making the soup. This is inefficient because you can’t do anything while waiting.

In contrast, the coroutine approach is like putting the rice in the rice cooker and, while waiting, preparing the ingredients for the stir-fry, or even starting to make the soup. When the rice cooker beeps, you go back to handle the cooked rice. This way, you can advance multiple tasks simultaneously, greatly improving efficiency.

In Python, coroutines are defined using async def:

import asyncio

# Define a coroutine function
async def fetch_data():
    print("Starting to fetch data...")
    # await indicates that this may require waiting, allowing control to be yielded
    await asyncio.sleep(2)  # Simulate IO operation
    print("Data fetching completed!")
    return "This is the fetched data"

Why use asyncio?

Performance Advantages

Traditional synchronous programs block the entire program when encountering IO operations, while asyncio can execute other tasks while waiting for IO, greatly enhancing the program’s concurrent processing capability.

Resource Savings

Compared to multithreading, asyncio does not require creating new threads for each task, avoiding the overhead of thread switching and memory consumption.

Avoiding Race Conditions

The single-threaded model naturally avoids race condition issues found in multithreaded programming, making the code safer and more reliable.

Suitable for IO-Bound Tasks

For IO-bound tasks such as network requests, file operations, and database queries, asyncio can deliver outstanding performance advantages.

How to use asyncio?

Basic Syntax

Defining Coroutine Functions

Use the async def keyword to define a coroutine function:

async def my_coroutine():
    print("This is a coroutine function")
    await asyncio.sleep(1)
    return "Execution completed"

Waiting for Coroutine Execution

Use the await keyword to wait for the coroutine to complete:

async def main():
    # Wait for the coroutine to complete
    result = await my_coroutine()
    print(result)

# Run the asynchronous program
asyncio.run(main())

Concurrent Execution of Multiple Coroutines

Use asyncio.gather() or asyncio.create_task() to execute multiple coroutines concurrently:

async def say_hello(name, delay):
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")

async def main():
    # Method 1: Use gather to execute concurrently
    await asyncio.gather(
        say_hello("Alice", 1),
        say_hello("Bob", 2), 
        say_hello("Charlie", 1.5)
    )
    
    # Method 2: Use create_task
    task1 = asyncio.create_task(say_hello("David", 1))
    task2 = asyncio.create_task(say_hello("Eve", 2))
    
    await task1
    await task2

asyncio.run(main())

Example

Using asyncio for concurrent file downloads:

import asyncio
import aiohttp
import aiofiles
from pathlib import Path

async def download_file(session, url, filename):
    """Asynchronously download a single file"""
    try:
        print(f"Starting download: {filename}")
        async with session.get(url) as response:
            if response.status == 200:
                # Use aiofiles to write the file asynchronously
                async with aiofiles.open(filename, 'wb') as f:
                    async for chunk in response.content.iter_chunked(1024):
                        await f.write(chunk)
                print(f"Download completed: {filename}")
            else:
                print(f"Download failed: {filename}, Status code: {response.status}")
    except Exception as e:
        print(f"Download error: {filename}, Error: {e}")

async def batch_download(download_list):
    """Batch download files"""
    # Create download directory
    download_dir = Path("downloads")
    download_dir.mkdir(exist_ok=True)
    
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url, filename in download_list:
            filepath = download_dir / filename
            task = download_file(session, url, filepath)
            tasks.append(task)
        
        # Execute all download tasks concurrently
        await asyncio.gather(*tasks)

# Usage example
download_list = [
    ("https://httpbin.org/uuid", "file1.json"),
    ("https://httpbin.org/json", "file2.json"), 
    ("https://httpbin.org/headers", "file3.json")
]

asyncio.run(batch_download(download_list))

Timeout Control

In practical applications, we often need to set a timeout for asynchronous operations:

import asyncio

async def long_running_task():
    """Simulate a time-consuming task"""
    await asyncio.sleep(5)
    return "Task completed"

async def main():
    try:
        # Set a 3-second timeout
        result = await asyncio.wait_for(long_running_task(), timeout=3.0)
        print(result)
    except asyncio.TimeoutError:
        print("Task timed out!")

asyncio.run(main())

Considerations

When to use asyncio

Suitable: IO-bound tasks such as network requests, file IO, and database operations
Not suitable: CPU-bound tasks (such as mathematical computations, image processing, etc.)

Avoid Blocking Operations

Avoid using synchronous blocking operations in asynchronous functions:

# ❌ Incorrect example
async def bad_example():
    time.sleep(1)  # This will block the entire event loop
    
# ✅ Correct example  
async def good_example():
    await asyncio.sleep(1)  # This will not block the event loop

We share Python programming knowledge every day, feel free to follow us! If you find it helpful, please like and share!