asyncio

asyncio Best Practices

asyncio is Python's standard library for writing concurrent, I/O-bound code using async/await. This guide provides opinionated, actionable rules to ensure your asyncio applications are structured, performant, and easy to maintain.

1. Structured Concurrency: Entry Points and Task Management

Always use asyncio.run() as the single top-level entry point for your asynchronous application. Manage concurrent operations explicitly with asyncio.create_task(), ensuring all tasks are awaited or handled.

1.1. Top-Level Entry Point

Use asyncio.run() once to start your main coroutine. Enable debug mode during development for critical warnings about un-awaited coroutines and resource leaks.

❌ BAD: Manually managing event loops or calling loop.run_until_complete().

import asyncio

async def main():
    print("Hello")

# Don't do this in application code; it's low-level and error-prone.
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

✅ GOOD: Use asyncio.run() with debug=True for development.

import asyncio

async def main():
    print("Hello from main!")
    await asyncio.sleep(0.1)
    print("Goodbye from main!")

if __name__ == "__main__":
    asyncio.run(main(), debug=True) # Always enable debug in dev!

1.2. Launching Concurrent Tasks

Use asyncio.create_task() to schedule coroutines to run concurrently. Always store a reference to the Task object to prevent it from being garbage collected prematurely, which can lead to silent failures and unhandled exceptions.

❌ BAD: Calling a coroutine without await or create_task().

import asyncio

async def fetch_data(url: str):
    print(f"Fetching {url}...")
    await asyncio.sleep(1) # Simulate network I/O
    print(f"Finished fetching {url}")
    return f"Data from {url}"

async def main():
    # This coroutine will never run!
    fetch_data("http://example.com/api/data")
    print("Main finished without waiting for data.")

if __name__ == "__main__":
    asyncio.run(main(), debug=True)
# Output: RuntimeWarning: coroutine 'fetch_data' was never awaited
# Main finished without waiting for data.