The Four Mistakes Everyone Makes
Blocking the loop, fire-and-forget tasks, calling a coroutine without await, and a gather whose exceptions nobody reads.
asyncio.run(main(), debug=True) # logs slow callbacks and un-awaited coroutinesAsync bugs rarely raise anything. They show up as latency, as work that silently did not happen, or as a log line nobody reads. These four account for most of them.
1. A synchronous call in a coroutine
time.sleep, requests.get, a psycopg2 query, a large read(): while any of them runs, no other task moves, no timer fires, no connection is accepted. asyncio.run(main(), debug=True) logs any callback slower than loop.slow_callback_duration (0.1s), which points straight at the culprit. The fix is an async driver, or await asyncio.to_thread(fn, *args).
2. Fire and forget
The loop keeps only weak references to tasks, so asyncio.create_task(send_email()) with the result thrown away can be garbage collected mid-execution. Keep a strong reference, or use a TaskGroup, which holds one for you:
background = set() task = asyncio.create_task(send_email()) background.add(task) task.add_done_callback(background.discard)
3. Mixing sync and async
Calling a coroutine function without await returns a coroutine object that is truthy, not falsy, and never runs — so if check_user(): ... is always true. And asyncio.run() inside a running loop raises RuntimeError; from async code, await instead.
4. A gather that hides the failure
gather(..., return_exceptions=True) never raises. If nobody inspects the returned list, every failure in the batch is invisible. Check each element against BaseException, or use a TaskGroup and let it raise.
Example
When to use it
- An async service whose p99 latency is terrible while CPU sits near zero.
- Emails or audit writes that are sent "sometimes" and nobody can reproduce the misses.
- A nightly batch that reports success while a tenth of its items failed.
- Reviewing a colleague's first async pull request and knowing what to look for.
More examples
Background tasks that survive to completion
Two bugs fixed at once: the set keeps the task from being garbage collected mid-flight, and the done callback retrieves the exception so a failure is logged at the moment it happens rather than whenever the object is collected. Use this when the work genuinely must outlive the request; prefer a TaskGroup whenever it does not.
import asyncio
import logging
_background = set()
def spawn(coro):
"""Fire-and-forget that actually finishes."""
task = asyncio.create_task(coro)
_background.add(task) # strong reference: not collectable
task.add_done_callback(_background.discard) # and not leaked, either
task.add_done_callback(_log_failure) # nothing awaits it, so log it here
return task
def _log_failure(task):
if not task.cancelled() and task.exception() is not None:
logging.exception("background task failed", exc_info=task.exception())
async def shutdown():
# on the way out, give them a chance to finish instead of dropping them
if _background:
await asyncio.wait(set(_background), timeout=5)The sync call you did not know was there
httpx.get and httpx.AsyncClient().get differ by one word and by every performance characteristic; the sync one inside a coroutine blocks the whole process and raises nothing. Grep an async codebase for requests., httpx.get, time.sleep and open( before trusting any latency measurement.
import asyncio
import json
from pathlib import Path
import httpx
async def handler_bad(path, url):
config = json.loads(Path(path).read_text()) # blocking disk read
r = httpx.get(url, timeout=5) # blocking HTTP: sync client!
return config, r.json()
async def handler_good(client, path, url):
# disk IO off the loop thread; one shared async client for HTTP
config = json.loads(await asyncio.to_thread(Path(path).read_text))
r = await client.get(url, timeout=5)
return config, r.json()
async def main():
async with httpx.AsyncClient() as client:
print(await handler_good(client, "config.json", "https://api.example.com/ping"))Making the failures visible again
return_exceptions=True converts exceptions into data, and data that nothing reads is the same as data that does not exist. Inspect the list, log with the identifier that produced each failure, and re-raise as an ExceptionGroup (3.11+) so the scheduler running the job sees a non-zero exit.
import asyncio
import logging
async def sync_all_bad(shops):
# looks fine, reports nothing: every failure is sitting in the list, unread
await asyncio.gather(*(sync(s) for s in shops), return_exceptions=True)
async def sync_all_ok(shops):
results = await asyncio.gather(*(sync(s) for s in shops), return_exceptions=True)
failures = [(s, r) for s, r in zip(shops, results) if isinstance(r, BaseException)]
for shop, exc in failures:
logging.error("sync failed for %s: %r", shop, exc)
if failures:
# one exception carrying all of them, so the job exits non-zero
raise ExceptionGroup("sync failed", [e for _, e in failures])
return results
Discussion