Debugging Asynchronous Python Code: Tools and Techniques

Introduction

Debugging asynchronous Python code can feel like solving a moving puzzle; while non-blocking execution improves performance, it also introduces race conditions, deadlocks, and unhandled exceptions. This article presents three essential tools that simplify async debugging with practical examples.

1. Tracking async bugs with pdb

The built-in pdb debugger, although not designed for async, can still inspect variables and step through code. A common mistake is forgetting to await a coroutine, causing it to never run. The example below demonstrates the issue and how inserting a breakpoint reveals the problem.

<code>import asyncio

async def fetch_data():
    print("Fetching data...")
    await asyncio.sleep(1)
    print("Data fetched.")

async def main():
    fetch_data()  # Forgot `await`!
    print("Task completed.")

asyncio.run(main())
</code>

Using pdb :

<code>import pdb

async def main():
    pdb.set_trace()  # Add breakpoint
    fetch_data()  # Missing `await`
    print("Task completed.")

asyncio.run(main())
</code>

Running the script pauses at the breakpoint, where inspecting the variable shows a coroutine object instead of its result. Adding the missing await resolves the bug.

2. Real-time event-loop inspection with aiomonitor

When async errors hide in the shadows—tasks freezing or overlapping— aiomonitor provides a live view of the event loop, listing active tasks and their states. After installing the package, wrap the main coroutine with aiomonitor.start_monitor() and connect via telnet.

<code>import asyncio

async def worker(name):
    for i in range(3):
        print(f"{name}: {i}")
        await asyncio.sleep(1)

async def main():
    task1 = asyncio.create_task(worker("Task1"))
    task2 = asyncio.create_task(worker("Task2"))
    await asyncio.gather(task1, task2)

asyncio.run(main())
</code>

Modified with aiomonitor :

<code>async def main():
    with aiomonitor.start_monitor():
        task1 = asyncio.create_task(worker("Task1"))
        task2 = asyncio.create_task(worker("Task2"))
        await asyncio.gather(task1, task2)

asyncio.run(main())
</code>

Connect to the monitor:

<code>telnet localhost 50101
</code>

In the REPL you can list all running tasks and investigate why a particular task is stuck.

3. Preventing bugs with asynctest

Debugging is only half the battle; preventing async bugs is crucial. The asynctest library lets you write tests that expose race conditions. The following example shows a typical race-condition failure and how adding an asyncio.Lock fixes it.

<code>import asyncio

counter = 0

async def increment():
    global counter
    temp = counter
    await asyncio.sleep(0.1)  # Simulate delay
    counter = temp + 1

async def main():
    await asyncio.gather(increment(), increment())

asyncio.run(main())
</code>

Test case with asynctest (will fail):

<code>import asynctest

class TestRaceCondition(asynctest.TestCase):
    async def test_race_condition(self):
        global counter
        counter = 0

        async def increment():
            global counter
            temp = counter
            await asyncio.sleep(0.1)
            counter = temp + 1

        await asyncio.gather(increment(), increment())

        self.assertEqual(counter, 2)  # This will fail
</code>

Solution using a lock:

<code>lock = asyncio.Lock()

async def increment():
    global counter
    async with lock:
        temp = counter
        await asyncio.sleep(0.1)
        counter = temp + 1
</code>

Rerunning the test now passes, confirming that the lock prevents overlapping increments.

Conclusion

Async Python debugging need not be a nightmare. Use pdb for simple missing‑await bugs, aiomonitor for live task inspection, and asynctest to catch and fix race conditions before they reach production.