Async Patterns for Python: Writing Modern, Concurrent Code#

Synchronous Approach#

In a typical synchronous (or “blocking”) program, each function call completes in its own time. If one function takes a while to finish—maybe it does network I/O or reads from a slow file—then everything else in the program waits until that function is done. This can cause a major bottleneck, particularly in I/O-heavy workflows.

Here’s a simple synchronous approach to reading two files, where each step blocks until completion:

1
def read_files_sync():
2
    with open("file1.txt", "r") as f1:
3
        data1 = f1.read()
4

5
    with open("file2.txt", "r") as f2:
6
        data2 = f2.read()
7

8
    print(len(data1), len(data2))

If file1.txt is huge or stored on a slow network volume, everything else just waits.

Asynchronous Approach#

Asynchronous code, on the other hand, attempts to avoid blocking. Instead of making a straight call to read a file, your program schedules an I/O operation and immediately returns control to the event loop, which can continue scheduling other tasks in parallel. When the data is ready (e.g., the I/O operation completes), the coroutine is resumed with the retrieved results.

An asynchronous approach to reading two files (simulated using async constructs) might look like this:

1
import asyncio
2
import aiofiles
3

4
async def read_file_async(file_path):
5
    async with aiofiles.open(file_path, mode='r') as f:
6
        return await f.read()
7

8
async def read_files_async():
9
    data1_task = asyncio.create_task(read_file_async("file1.txt"))
10
    data2_task = asyncio.create_task(read_file_async("file2.txt"))
11

12
    # Wait for both tasks to finish
13
    data1, data2 = await asyncio.gather(data1_task, data2_task)
14

15
    print(len(data1), len(data2))
16

17
asyncio.run(read_files_async())

By using async I/O, we can read from multiple files concurrently, ensuring that idle time waiting for disk operations (or network I/O) is spent doing other tasks.

Basic Terms#

• Coroutine: A specialized function defined with async def. It can be suspended and resumed via await.
• Task: A scheduled coroutine. You can create a task using asyncio.create_task(coro).
• Event Loop: The orchestrator that runs tasks, handles callbacks, and manages I/O in a single-threaded, event-driven manner.

Minimal Asyncio Example#

1
import asyncio
2

3
async def hello():
4
    print("Hello Async World!")
5

6
async def main():
7
    # Create and schedule the task
8
    task = asyncio.create_task(hello())
9
    # Wait for the task to complete
10
    await task
11

12
asyncio.run(main())

Here’s the breakdown:

1. hello() is an async coroutine that prints a message.
2. main() schedules hello() using asyncio.create_task(hello()) and then awaits its completion.
3. asyncio.run(main()) enters the event loop to execute main.

By structuring functions as coroutines and awaiting them, we compose asynchronous flows that can overlappingly perform I/O.

Event Loop Lifecycle#

• Start: When you call asyncio.run(...), Python sets up a new event loop (unless one is already running), initializes tasks, and begins processing events.
• Execution: Each task runs in small increments, yielding control via await whenever an I/O operation is requested. While that operation is in flight, the event loop can schedule another ready task, maximizing concurrency.
• Stop: If the event loop detects there are no more tasks left to run, it shuts down gracefully.

Because asyncio.run() sets up and tears down the loop each time, you typically only call it once at the top-level script entry point. Inside your code, you rarely need to manage the event loop directly—asyncio does that for you.

Code Snippet Demonstrating Tasks#

1
import asyncio
2

3
async def compute_square(n):
4
    await asyncio.sleep(1)  # Simulate I/O delay
5
    return n * n
6

7
async def main():
8
    # Create multiple tasks
9
    tasks = [
10
        asyncio.create_task(compute_square(i))
11
        for i in range(1, 6)
12
    ]
13
    # Gather results
14
    results = await asyncio.gather(*tasks)
15
    print("Squares:", results)
16

17
asyncio.run(main())

Here, we generate 5 separate tasks that each compute a square after a simulated 1-second delay. Because of concurrency, the total runtime will be roughly 1 second (not 5), as each I/O suspension can overlap.

Example: Basic TCP Echo Server#

Below is a minimal example of a TCP echo server using asyncio. It listens for incoming connections, reads data, and sends it back:

1
import asyncio
2

3
async def handle_echo(reader, writer):
4
    data = await reader.read(1024)
5
    message = data.decode()
6
    addr = writer.get_extra_info('peername')
7
    print(f"Received {message!r} from {addr}")
8

9
    writer.write(data)
10
    await writer.drain()
11

12
    print("Close the connection")
13
    writer.close()
14
    await writer.wait_closed()
15

16
async def main():
17
    server = await asyncio.start_server(
18
        handle_echo, '127.0.0.1', 8888)
19

20
    addr = server.sockets[0].getsockname()
21
    print(f"Serving on {addr}")
22

23
    async with server:
24
        await server.serve_forever()
25

26
asyncio.run(main())

Notice the async with server: await server.serve_forever() pattern. This keeps the server running until signaled to stop while using async calls for socket operations.

1. Gathering Tasks#

asyncio.gather() is a straightforward way to launch multiple coroutines concurrently and wait for their results:

1
results = await asyncio.gather(
2
    coro1(),
3
    coro2(),
4
    coro3()
5
)

It returns a list/tuple of results in the order the coroutines were passed to gather. Any exception in a single coroutine will cancel the others unless you specify additional parameters (like return_exceptions=True).

2. Race Conditions#

Sometimes you want the first task that completes, ignoring the rest. Python 3.11+ introduced asyncio.TaskGroup, but you can also do something akin to a “race” using lower-level constructs. For instance, you might gather tasks in a group and cancel the ones you no longer need once one completes. Alternatively, you can use third-party libraries that offer a race function akin to JavaScript’s Promise.race().

3. Producer-Consumer Pattern#

This pattern often involves one or more coroutines producing data (e.g., reading from a sensor, fetching from an API) and other coroutines consuming that data (e.g., writing to a database, analyzing). Typically, an asyncio.Queue helps implement this cleanly:

1
import asyncio
2

3
async def producer(queue):
4
    for i in range(5):
5
        print(f"Producing item {i}")
6
        await queue.put(i)
7
        await asyncio.sleep(1)
8

9
async def consumer(queue):
10
    while True:
11
        item = await queue.get()
12
        if item is None:
13
            break
14
        print(f"Consuming item {item}")
15
        queue.task_done()
16

17
async def main():
18
    q = asyncio.Queue()
19
    consumer_task = asyncio.create_task(consumer(q))
20
    await producer(q)
21

22
    # Signal the consumer to exit
23
    await q.put(None)
24

25
    # Wait for consumer to finish
26
    await consumer_task
27

28
asyncio.run(main())

This design is robust and helps decouple the production rate from the consumption rate, enabling concurrency in a pipeline-like fashion.

4. Cancellation and Timeouts#

In real-world scenarios, you often need to handle cancellations and timeouts. Suppose you’re calling an external API or a slow operation, you might wrap it in asyncio.wait_for(coro, timeout) to ensure your coroutine doesn’t hang indefinitely:

1
import asyncio
2

3
async def slow_operation():
4
    await asyncio.sleep(5)
5
    return "Done"
6

7
async def main():
8
    try:
9
        result = await asyncio.wait_for(slow_operation(), timeout=2)
10
        print(result)
11
    except asyncio.TimeoutError:
12
        print("Operation took too long!")
13

14
asyncio.run(main())

Handling cancellations is also crucial. If a user presses Ctrl+C or if you explicitly cancel a task, your coroutine should properly clean up resources (database connections, file handles, partial writes) before shutting down.

Summary Table of Common Pitfalls#

1. Testing Async Functions with pytest#

Pytest supports async code testing with the pytest-asyncio plugin. You can mark your test functions as async:

1
import pytest
2
import asyncio
3

4
@pytest.mark.asyncio
5
async def test_example():
6
    async def async_add(x, y):
7
        await asyncio.sleep(0.1)
8
        return x + y
9

10
    result = await async_add(2, 3)
11
    assert result == 5

The plugin arranges an event loop for each test, ensuring your coroutines run properly.

2. Debug Mode#

Set the event loop to debug mode to track tasks:

1
import asyncio
2

3
async def main():
4
    # Your async code here
5
    pass
6

7
if __name__ == "__main__":
8
    asyncio.run(main(), debug=True)

Debug mode improves logging, detects slow callbacks, warns about missed awaits, and can catch resource leaks more easily.

3. Logging#

When dealing with concurrency, standard print statements might get confusing as multiple tasks produce output interleaved. Instead, use Python’s logging module and ensure each coroutine logs meaningful context (e.g., request ID, coroutine ID, etc.):

1
import asyncio
2
import logging
3

4
logging.basicConfig(level=logging.INFO)
5
logger = logging.getLogger(__name__)
6

7
async def do_work(task_name):
8
    logger.info(f"{task_name} starting...")
9
    await asyncio.sleep(1)
10
    logger.info(f"{task_name} done!")
11

12
async def main():
13
    tasks = [do_work(f"Task-{i}") for i in range(3)]
14
    await asyncio.gather(*tasks)
15

16
asyncio.run(main())

This approach helps isolate which task is doing what.

1. Async Context Managers#

Python supports asynchronous context managers, which help manage resources that must be cleaned up properly. For example, an async database session might use:

1
class AsyncDBConnection:
2
    async def __aenter__(self):
3
        self.conn = await async_connect_to_db()
4
        return self.conn
5

6
    async def __aexit__(self, exc_type, exc, tb):
7
        await self.conn.close()
8

9
async def main():
10
    async with AsyncDBConnection() as conn:
11
        data = await conn.query("SELECT * FROM table")
12
        print(data)

2. TaskGroups (Python 3.11+)#

Python 3.11 introduced asyncio.TaskGroup, providing a structured concurrency primitive similar to nursery blocks in the Trio library. It ensures all tasks within a task group run together, and if one fails, the rest get canceled. Example:

1
import asyncio
2

3
async def say_hello(name):
4
    await asyncio.sleep(1)
5
    print(f"Hello, {name}")
6

7
async def main():
8
    async with asyncio.TaskGroup() as tg:
9
        tg.create_task(say_hello("Alice"))
10
        tg.create_task(say_hello("Bob"))
11

12
asyncio.run(main())

3. Other Frameworks#

• Trio: A popular alternative to asyncio that simplifies structured concurrency. Trio heavily emphasizes correctness and approachability.
• curio: Another library that explores coroutines with an emphasis on system-level concurrency.
• gevent: Uses greenlets and monkey patching to achieve concurrency. Common in older codebases but less modern than native asyncio.

Each framework has its own strengths, but asyncio remains the standard for most contemporary Python async development.

1. Web Scraping#

Async is ideal for sending many simultaneous HTTP requests without blocking. Libraries like aiohttp let you scale web scraping or microservices:

1
import asyncio
2
import aiohttp
3

4
async def fetch(session, url):
5
    async with session.get(url) as response:
6
        return await response.text()
7

8
async def scrape_sites(urls):
9
    async with aiohttp.ClientSession() as session:
10
        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
11
        results = await asyncio.gather(*tasks)
12
        return results
13

14
urls = ["http://example.com", "http://python.org", "http://github.com"]
15
all_data = asyncio.run(scrape_sites(urls))
16
print(len(all_data))

2. Chat Server#

Building real-time applications like chat servers or multiplayer games naturally benefits from async concurrency. These applications typically must handle simultaneous connections, broadcast updates, and manage state without blocking.

3. Batch Processing#

Even when performing tasks on a local machine—such as reading multiple big files, interacting with local databases, or working with external APIs—async patterns can help pipeline these operations efficiently.

4. Microservices#

In a microservices architecture, you might have services that continuously request data from other services. Async design ensures your service doesn’t spend time idle, waiting for responses.