3 Essential Tools You Must Learn for Debugging Async Code in Python
Debugging async code in Python can feel like solving a puzzle where the pieces keep moving. The non-blocking nature of asyncio
is great for performance but introduces challenges like race conditions, deadlocks, and unhandled exceptions. Over the years, I’ve relied on three key tools—pdb
, aiomonitor
, and asynctest
—to make debugging async Python code manageable. In this article, I’ll share practical examples of how these tools can help you debug effectively and save countless hours.
1. Use pdb
to Track Down Async Bugs
If you’ve ever debugged Python code, you’ve probably used pdb
, Python’s built-in debugger. While it’s not tailored for async workflows, it’s still a reliable tool for inspecting variables and stepping through code.
A Bug That’s Hard to Spot
Here’s a simple async program with an easy-to-miss mistake:
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())
The issue? The program finishes without errors, but fetch_data()
doesn’t run. This happens because fetch_data()
was called but not awaited, so it didn’t execute.
Debugging It with pdb
To catch the problem, add a breakpoint with pdb
:
import pdb
async def main():
pdb.set_trace() # Add breakpoint
fetch_data() # Missing `await`
print("Task completed.")
asyncio.run(main())
Run the script, and when it pauses:
- Step into the code using the
s
command. - Type
fetch_data()
to inspect its value. You’ll see it’s a coroutine object, not the result.
Once you spot the issue, fix it by adding await
:
async def main():
await fetch_data()
print("Task completed.")
asyncio.run(main())
pdb
is great for catching simple bugs like this, but for more complex issues, we need specialized tools.
2. Use aiomonitor
for Real-Time Event Loop Debugging
When async bugs feel like they’re hiding in the shadows—tasks freezing or overlapping in ways you can’t reproduce—aiomonitor
can save the day. It lets you inspect the event loop in real time, showing active tasks and their states.
Debugging Stuck Tasks
Here’s a program where two tasks run concurrently, but one occasionally freezes:
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())
When one of the workers stops printing, it’s time to bring in aiomonitor
.
Debugging with aiomonitor
Install aiomonitor
:
pip install aiomonitor
Modify the code to include the monitor:
import aiomonitor
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())
Run the program and connect to the monitor using Telnet:
telnet localhost 50101
In the REPL, type tasks
to see all running tasks. Use task <id>
to inspect specific tasks and identify why one might be stuck or waiting.
aiomonitor
shines when you need real-time visibility into your event loop and running tasks.
3. Use asynctest
to Prevent Bugs from Sneaking In
Debugging is only half the battle. The best way to handle async bugs is to prevent them, and that’s where asynctest
comes in. It’s a testing library built specifically for async Python code, making it easy to mock coroutines and write test cases.
Testing a Race Condition
Here’s a classic race condition bug:
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())
After running this code, counter
might end up as 1
instead of 2
. Let’s catch this bug using asynctest
.
Writing a Test with asynctest
Install asynctest
:
pip install asynctest
Write a test case to expose the race condition:
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
Fixing the Issue
Add a lock to prevent overlapping increments:
lock = asyncio.Lock()
async def increment():
global counter
async with lock:
temp = counter
await asyncio.sleep(0.1)
counter = temp + 1
Run the test again, and it will pass.
Wrapping It Up
Debugging async Python code doesn’t have to be a nightmare. With the right tools:
pdb
helps you track down simple issues like unawaited coroutines.aiomonitor
gives you real-time insights into running tasks and the event loop.asynctest
ensures bugs are caught and fixed before they ever reach production.
Each of these tools has saved me countless hours—and a fair share of headaches—when working on async projects. I hope they’ll do the same for you. Happy debugging!