Introduction

In Python asyncio, if you create a task with asyncio.create_task() or schedule a coroutine but never await it, any exception raised inside that coroutine is silently discarded. The task completes with an exception, but nobody checks for it.

This is one of the most insidious bugs in async Python because the code appears to work -- it just silently skips the failed operation with no error message.

Symptoms

  • Async code runs without errors but expected side effects do not occur
  • Database writes, file operations, or API calls silently fail
  • No exception traceback appears in logs despite faulty async code

Common Causes

  • Calling an async function without await: fetch_data() instead of await fetch_data()
  • Using asyncio.create_task() without storing and awaiting the task
  • Exception in background task not retrieved before the task is garbage collected

Step-by-Step Fix

  1. 1.Enable asyncio debug mode to detect unawaited coroutines: Python 3.12+ warns about unawaited coroutines.
  2. 2.```python
  3. 3.import asyncio
  4. 4.import os

# Enable debug mode os.environ['PYTHONASYNCIODEBUG'] = '1' asyncio.run(main(), debug=True)

# Output warning: # RuntimeWarning: coroutine 'fetch_data' was never awaited # fetch_data() # ~~~~~~~~~~^^ ```

  1. 1.Always await or gather tasks with exception handling: Use asyncio.gather with return_exceptions to catch all errors.
  2. 2.```python
  3. 3.import asyncio

async def fetch_data(url): raise ConnectionError(f"Failed to connect to {url}")

async def main(): urls = ['http://a.com', 'http://b.com', 'http://c.com']

# GOOD: catch all exceptions results = await asyncio.gather( *[fetch_data(u) for u in urls], return_exceptions=True ) for url, result in zip(urls, results): if isinstance(result, Exception): print(f"Error fetching {url}: {result}") else: print(f"Got result: {result}")

asyncio.run(main()) ```

  1. 1.Add task exception callbacks for background tasks: Use add_done_callback to log exceptions from fire-and-forget tasks.
  2. 2.```python
  3. 3.import asyncio
  4. 4.import logging

logger = logging.getLogger(__name__)

def log_task_exception(task): if task.exception() is not None: logger.error(f"Task {task.get_name()} failed: {task.exception()}")

async def main(): task = asyncio.create_task(background_worker()) task.set_name('background_worker') task.add_done_callback(log_task_exception)

asyncio.run(main()) ```

  1. 1.Use asyncio.wait with FIRST_EXCEPTION: Stop execution when any task fails.
  2. 2.```python
  3. 3.import asyncio

async def main(): tasks = { asyncio.create_task(fetch_api()), asyncio.create_task(fetch_db()), asyncio.create_task(fetch_cache()), }

done, pending = await asyncio.wait( tasks, return_when=asyncio.FIRST_EXCEPTION )

for task in done: if task.exception(): print(f"Task failed: {task.exception()}") for p in pending: p.cancel() break

asyncio.run(main()) ```

Prevention

  • Enable asyncio debug mode in development: asyncio.run(main(), debug=True)
  • Use return_exceptions=True with asyncio.gather in production code
  • Add done callbacks to all fire-and-forget background tasks
  • Run async linters like flake8-async or ruff with async rules enabled