30 Async Rules

Q: async sqlalchemy pattern?

async with async_session() as session: per request via FastAPI dependency.

Q: async generators?

async for with async generators. Close with aclose() in finally blocks.

Thirty rules for writing correct, performant asyncio code in Python 3.14. Violating these causes subtle bugs that only appear under load.

Recipe

Review when adding async def to any module. Pair with pytest-asyncio tests.

When to reach for this:

Building FastAPI services or async workers
Migrating sync code to asyncio
Debugging event-loop freezes or task leaks

Deep Dive

Event Loop Fundamentals (1-10)

One event loop per thread - never call asyncio.run() inside a running loop.
Use asyncio.run() as main entry - top-level script and sync test runner only.
Prefer asyncio.TaskGroup (3.14) - structured concurrency over bare create_task.
Await every coroutine - unawaited coroutines emit RuntimeWarning and do nothing.
Do not mix sync and async APIs - requests in async def blocks the loop.
Use asyncio.to_thread() for blocking I/O - when no async library exists.
Install pytest-asyncio - asyncio_mode = "auto" in pyproject.toml.
Async fixtures for async setup - database connections, httpx clients.
Cancel tasks on shutdown - handle asyncio.CancelledError gracefully.
Set timeouts on network calls - asyncio.wait_for() or httpx timeout=.

Concurrency Patterns (11-20)

asyncio for I/O, not CPU - CPU work goes to ProcessPoolExecutor.
Cap concurrency with Semaphore - async with sem: limits parallel requests.
Use asyncio.gather for independent tasks - return_exceptions=True for partial failure.
Use asyncio.Queue for producer-consumer - bounded queues prevent memory blowup.
Avoid shared mutable state - pass data via arguments; use locks if shared.
Connection pools are async too - one AsyncClient or pool per app, not per request.
Lazy-init async resources in lifespan - FastAPI lifespan or @app.on_event startup.
Background tasks must handle errors - log exceptions in tasks; do not fire-and-forget silently.
Prefer async context managers - async with session.begin(): for transaction boundaries.
Stream large responses - StreamingResponse instead of loading full body in memory.

Pitfalls & Debugging (21-30)

Never time.sleep() in async - use await asyncio.sleep().
Beware @lru_cache on async functions - cache coroutine objects, not results. Use async_lru.
DNS is blocking - use aiodns or run resolver in thread for high-concurrency apps.
SSL handshake is blocking - use async HTTP clients (httpx, aiohttp), not urllib.
Watch for task leaks - track background tasks; cancel on request completion.
Test cancellation paths - client disconnect should cancel server work.
Log task exceptions - task.add_done_callback to log unhandled errors.
Do not reuse closed loops - create fresh loop per test or use pytest-asyncio.
Free-threaded 3.14 needs audit - shared state without GIL requires locks.
Profile under concurrent load - single-request profiling misses contention bugs.

Gotchas

Blocking call in async route - entire server stalls. Fix: await asyncio.to_thread(blocking_fn).
create_task without reference - task garbage-collected mid-execution. Fix: store task reference or use TaskGroup.
asyncio.gather without error handling - one failure cancels all. Fix: return_exceptions=True or explicit try/except per task.
Global async client - stale connections after fork. Fix: create per-process in lifespan; close on shutdown.
Testing with asyncio.run in each test - slow and loop conflicts. Fix: pytest-asyncio with auto mode.

Alternatives

Alternative	Use When	Don't Use When
threading	Blocking libraries only	High concurrency I/O
multiprocessing	CPU-bound parallelism	Shared state needed
Sync FastAPI + workers	Team unfamiliar with async	I/O-heavy with async drivers available

FAQs

When should I use async?

When the app is I/O-bound and async libraries exist for your database, HTTP, and cache.

FastAPI must be async?

No. Sync routes work via thread pool. Async routes preferred with async DB drivers.

TaskGroup vs gather?

TaskGroup (3.11+) for structured concurrency with automatic cancellation on failure.

How do I debug a frozen server?

py-spy on the process. Look for sync blocking in async call stacks.

async sqlalchemy pattern?

async with async_session() as session: per request via FastAPI dependency.

Can Django be async?

Django 5.x supports async views. ORM async support is partial; evaluate per use case.

How many concurrent tasks?

Start with semaphore of 10-50. Tune based on connection limits and memory.

What about trio?

anyio bridges asyncio and trio. FastAPI ecosystem is asyncio-native.

async generators?

async for with async generators. Close with aclose() in finally blocks.

How do I migrate sync to async?

One endpoint at a time. Swap sync driver for async driver. Test under load.

Async & Concurrency for Throughput - performance patterns
Testing Async Code - async tests
Concurrency - GIL and models
40 API Design Rules - async API rules

Stack versions: This page was written for Python 3.14.0, FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.

Recipe

Review when adding async def to any module. Pair with pytest-asyncio tests.

When to reach for this:

Building FastAPI services or async workers
Migrating sync code to asyncio
Debugging event-loop freezes or task leaks

Deep Dive

Event Loop Fundamentals (1-10)

One event loop per thread - never call asyncio.run() inside a running loop.
Use asyncio.run() as main entry - top-level script and sync test runner only.
Prefer asyncio.TaskGroup (3.14) - structured concurrency over bare create_task.
Await every coroutine - unawaited coroutines emit RuntimeWarning and do nothing.
Do not mix sync and async APIs - requests in async def blocks the loop.
Use asyncio.to_thread() for blocking I/O - when no async library exists.
Install pytest-asyncio - asyncio_mode = "auto" in pyproject.toml.
Async fixtures for async setup - database connections, httpx clients.
Cancel tasks on shutdown - handle asyncio.CancelledError gracefully.
Set timeouts on network calls - asyncio.wait_for() or httpx timeout=.

Concurrency Patterns (11-20)

asyncio for I/O, not CPU - CPU work goes to ProcessPoolExecutor.
Cap concurrency with Semaphore - async with sem: limits parallel requests.
Use asyncio.gather for independent tasks - return_exceptions=True for partial failure.
Use asyncio.Queue for producer-consumer - bounded queues prevent memory blowup.
Avoid shared mutable state - pass data via arguments; use locks if shared.
Connection pools are async too - one AsyncClient or pool per app, not per request.
Lazy-init async resources in lifespan - FastAPI lifespan or @app.on_event startup.
Background tasks must handle errors - log exceptions in tasks; do not fire-and-forget silently.
Prefer async context managers - async with session.begin(): for transaction boundaries.
Stream large responses - StreamingResponse instead of loading full body in memory.

Pitfalls & Debugging (21-30)

Never time.sleep() in async - use await asyncio.sleep().
Beware @lru_cache on async functions - cache coroutine objects, not results. Use async_lru.
DNS is blocking - use aiodns or run resolver in thread for high-concurrency apps.
SSL handshake is blocking - use async HTTP clients (httpx, aiohttp), not urllib.
Watch for task leaks - track background tasks; cancel on request completion.
Test cancellation paths - client disconnect should cancel server work.
Log task exceptions - task.add_done_callback to log unhandled errors.
Do not reuse closed loops - create fresh loop per test or use pytest-asyncio.
Free-threaded 3.14 needs audit - shared state without GIL requires locks.
Profile under concurrent load - single-request profiling misses contention bugs.

Gotchas

Blocking call in async route - entire server stalls. Fix: await asyncio.to_thread(blocking_fn).
create_task without reference - task garbage-collected mid-execution. Fix: store task reference or use TaskGroup.
asyncio.gather without error handling - one failure cancels all. Fix: return_exceptions=True or explicit try/except per task.
Global async client - stale connections after fork. Fix: create per-process in lifespan; close on shutdown.
Testing with asyncio.run in each test - slow and loop conflicts. Fix: pytest-asyncio with auto mode.

Alternatives

Alternative	Use When	Don't Use When
threading	Blocking libraries only	High concurrency I/O
multiprocessing	CPU-bound parallelism	Shared state needed
Sync FastAPI + workers	Team unfamiliar with async	I/O-heavy with async drivers available

FAQs

When should I use async?

When the app is I/O-bound and async libraries exist for your database, HTTP, and cache.

FastAPI must be async?

No. Sync routes work via thread pool. Async routes preferred with async DB drivers.

TaskGroup vs gather?

TaskGroup (3.11+) for structured concurrency with automatic cancellation on failure.

How do I debug a frozen server?

py-spy on the process. Look for sync blocking in async call stacks.

async sqlalchemy pattern?

async with async_session() as session: per request via FastAPI dependency.

Can Django be async?

Django 5.x supports async views. ORM async support is partial; evaluate per use case.

How many concurrent tasks?

Start with semaphore of 10-50. Tune based on connection limits and memory.

What about trio?

anyio bridges asyncio and trio. FastAPI ecosystem is asyncio-native.

async generators?

async for with async generators. Close with aclose() in finally blocks.

How do I migrate sync to async?

One endpoint at a time. Swap sync driver for async driver. Test under load.

Async & Concurrency for Throughput - performance patterns
Testing Async Code - async tests
Concurrency - GIL and models
40 API Design Rules - async API rules

Stack versions: This page was written for Python 3.14.0, FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.

30 Async Rules

Search Documentation

30 Async Rules