Async Python’s bugs look different from sync. Stack traces don’t span async boundaries cleanly; blocking calls silently kill performance; tasks leak invisibly. This post is the working set of debugging patterns.
Enable debug mode
PYTHONASYNCIODEBUG=1 python myapp.py
Or:
import asyncio
asyncio.get_event_loop().set_debug(True)
What it catches:
- Slow callbacks (>100ms blocking the loop).
- Coroutines created but not awaited.
- Resources not closed.
- Tasks destroyed while pending.
Run integration tests with debug mode on. Most asyncio bugs surface here.
Detect blocking calls
The classic asyncio bug:
async def fetch():
requests.get("...") # ⛔ sync call; blocks event loop
Debug mode logs:
WARNING - Executing <Task pending coro=<...>> took 5.234 seconds
Once you see “executing took >1s”, you have a blocking call. Find it; replace with async equivalent (httpx.AsyncClient).
Find task leaks
import asyncio
async def diagnose():
while True:
await asyncio.sleep(60)
tasks = asyncio.all_tasks()
print(f"active tasks: {len(tasks)}")
for t in list(tasks)[:5]:
print(f" {t.get_name()}: {t.get_coro()}")
If task count grows without bound: leak. Usually fire-and-forget without tracking:
asyncio.create_task(some_work()) # ⚠ no reference; might be GC'd or leak
Fix: keep references; cancel them in shutdown.
Profile
import cProfile, pstats
import asyncio
async def main():
# your app
pass
profiler = cProfile.Profile()
profiler.enable()
asyncio.run(main())
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(30)
Same as sync profiling; works for async with caveat that “time” is wall time, including I/O waits.
For better async-aware profiling: py-spy (py-spy record -o profile.svg --pid <pid>) — sampling profiler that doesn’t require code changes.
Trace slow operations
import asyncio, time, contextlib
@contextlib.asynccontextmanager
async def timed(name):
start = time.perf_counter()
try:
yield
finally:
ms = (time.perf_counter() - start) * 1000
if ms > 100:
print(f"slow: {name} took {ms:.0f}ms")
async def fetch_user(id):
async with timed("fetch_user.db"):
await db.fetchrow(...)
async with timed("fetch_user.cache"):
await redis.get(...)
Lightweight tracing for slow operations. Combined with OpenTelemetry , you get production traces.
Detect cancellation issues
A common bug: catching CancelledError:
try:
await something()
except Exception:
pass # ⛔ swallows cancellation
Always re-raise:
try:
await something()
except CancelledError:
raise # explicit
except Exception as e:
log(e)
Or use Exception (which doesn’t catch BaseException / CancelledError).
Heartbeat detection
For long-running services, a periodic heartbeat task:
async def heartbeat():
while True:
await asyncio.sleep(5)
log(f"heartbeat: {time.time():.0f}, tasks={len(asyncio.all_tasks())}")
If heartbeat stops logging in an otherwise-running process, the loop is blocked.
Common bugs
1. await missed
result = some_coro() # ⛔ creates coroutine; never runs
result = await some_coro() # ✅
Debug mode warns. Linters like ruff catch many.
2. Sync DB driver in async code
psycopg2 instead of asyncpg. Each query blocks the loop. Performance dies.
3. asyncio.gather swallowing errors
results = await asyncio.gather(a(), b(), c()) # one fails; gather raises; others silently still running?
Modern: use TaskGroup .
4. Forgetting to close resources
DB connections, HTTP clients, files. Always async with or explicit close.
5. Long-held locks across await
async with lock:
result = await slow_call() # holds lock for the slow call
Either lock briefly + extract data + release + slow_call, or accept the contention.
Read this next
- Modern AsyncIO Patterns — TaskGroup, anyio
- Background Jobs in Python
- FastAPI WebSockets and Real-Time
- OpenTelemetry End-to-End
If you want my asyncio diagnostic helpers (timers, leak detector, blocking-call sniffer), it’s at rajpoot.dev .
Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .