asyncio

The skill Copy

# 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()`.
```python
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.
```python
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()`.
```python
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.
```

✅ GOOD: Explicitly create and await tasks.
```python
import asyncio
from typing import Awaitable

async def fetch_data(url: str) -> str:
    print(f"Fetching {url}...")
    await asyncio.sleep(1)
    print(f"Finished fetching {url}")
    return f"Data from {url}"

async def main() -> None:
    task: Awaitable[str] = asyncio.create_task(fetch_data("http://example.com/api/data"))
    print("Main continues while data is fetched...")
    result: str = await task # Await the task to get its result
    print(f"Received: {result}")

if __name__ == "__main__":
    asyncio.run(main(), debug=True)
```

### 1.3. Grouping Tasks

For running multiple tasks concurrently and waiting for all of them, use `asyncio.gather()`. For more fine-grained control over completion conditions (e.g., first to complete), use `asyncio.wait()`.

❌ BAD: Awaiting tasks sequentially when they can run in parallel.
```python
import asyncio
import time

async def process_item(item_id: int):
    print(f"Processing item {item_id}...")
    await asyncio.sleep(1)
    print(f"Item {item_id} processed.")
    return f"Result {item_id}"

async def main():
    start = time.monotonic()
    # This will take ~3 seconds total
    await process_item(1)
    await process_item(2)
    await process_item(3)
    print(f"Total time: {time.monotonic() - start:.2f}s")

if __name__ == "__main__":
    asyncio.run(main(), debug=True)
```

✅ GOOD: Use `asyncio.gather()` for concurrent execution.
```python
import asyncio
import time
from typing import List

async def process_item(item_id: int) -> str:
    print(f"Processing item {item_id}...")
    await asyncio.sleep(1)
    print(f"Item {item_id} processed.")
    return f"Result {item_id}"

async def main() -> None:
    start = time.monotonic()
    # This will take ~1 second total
    results: List[str] = await asyncio.gather(
        process_item(1),
        process_item(2),
        process_item(3)
    )
    print(f"All items processed. Results: {results}")
    print(f"Total time: {time.monotonic() - start:.2f}s")

if __name__ == "__main__":
    asyncio.run(main(), debug=True)
```

## 2. Avoiding Blocking Operations

The `asyncio` event loop must *never* be blocked. Any CPU-bound work or synchronous I/O operations must be offloaded to an executor.

### 2.1. CPU-Bound Work

Use `loop.run_in_executor()` with a `ThreadPoolExecutor` (for CPU-bound work in a separate thread) or `ProcessPoolExecutor` (for true parallelism, avoiding GIL) to prevent blocking the event loop.

❌ BAD: Performing heavy computation directly in a coroutine.
```python
import asyncio
import time

async def cpu_intensive_task():
    print("Starting CPU-intensive task...")
    # Simulate heavy computation - this blocks the event loop!
    _ = sum(i * i for i in range(10**7))
    print("Finished CPU-intensive task.")
    return "CPU result"

async def main():
    start = time.monotonic()
    task = asyncio.create_task(cpu_intensive_task())
    await asyncio.sleep(0.1) # This sleep will be delayed by the CPU task
    print("Main continued (but was blocked).")
    result = await task
    end = time.monotonic()
    print(f"Total time: {end - start:.2f}s, Result: {result}")

if __name__ == "__main__":
    asyncio.run(main(), debug=True)
# Output shows sleep was blocked, total time > 1s
```

✅ GOOD: Offload CPU-bound work to `run_in_executor`.
```python
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_cpu_task() -> str:
    print("Starting blocking CPU task in executor...")
    _ = sum(i * i for i in range(10**7))
    print("Finished blocking CPU task in executor.")
    return "CPU result"

async def main() -> None:
    start = time.monotonic()
    # Use default ThreadPoolExecutor (None) or a custom one
    result_future = asyncio.get_running_loop().run_in_executor(
        None, blocking_cpu_task
    )
    await asyncio.sleep(0.1) # This sleep will run concurrently
    print("Main continued

How to install it Copy

1. Open the tool and start a new conversation, project or template.
2. Paste the text below wherever the tool keeps standing instructions
   — usually called system prompt, instructions, or persona.
3. Save it so it applies to every run rather than a single message.

Did it work for you?

Where it came from