Asyncio Basics
9 examples to get you started with Asyncio - 6 basic and 3 intermediate.
Prerequisites
- Python 3.14.0 with
asynciostdlib. - Use
async/awaitsyntax - not legacy@asyncio.coroutinegenerators.
Basic Examples
1. Coroutine Definition
Coroutines are async functions paused at await.
import asyncio
async def hello() -> str:
await asyncio.sleep(0.1)
return "hello"
async def main() -> None:
print(await hello())
asyncio.run(main())async defdefines a coroutine function.- Calling it returns a coroutine object - must be awaited or scheduled.
asyncio.runmanages loop lifecycle for scripts.
Related: Tasks & Gather - concurrent coroutines
2. await Pauses Cooperatively
Other tasks run while one awaits I/O.
import asyncio
async def tick(name: str) -> None:
print(name, "start")
await asyncio.sleep(0.1)
print(name, "end")
async def main() -> None:
await tick("a")
await tick("b")
asyncio.run(main())- Sequential await runs one after another.
- Sleep yields control to the event loop.
- No parallelism on CPU - cooperative scheduling only.
Related: Debugging Async Code - blocking mistakes
3. gather for Concurrency
Run multiple coroutines concurrently.
import asyncio
async def work(n: int) -> int:
await asyncio.sleep(0.1)
return n
async def main() -> None:
results = await asyncio.gather(work(1), work(2), work(3))
print(results)
asyncio.run(main())gatherschedules all awaitables on the same loop.- Returns list in submission order when all succeed.
- One failure cancels others unless
return_exceptions=True.
Related: Tasks & Gather - Task objects
4. create_task Fire-and-Forget Scheduling
Schedule background work without awaiting immediately.
import asyncio
async def background() -> None:
await asyncio.sleep(0.1)
print("done")
async def main() -> None:
task = asyncio.create_task(background())
print("scheduled")
await task
asyncio.run(main())create_taskwraps coroutine in a Task.- Tasks run concurrently until awaited or loop stops.
- Store task references to avoid GC mid-flight (rare edge case).
Related: Structured Concurrency - TaskGroup
5. asyncio.sleep vs time.sleep
Never block the loop with sync sleep.
import asyncio
async def good() -> None:
await asyncio.sleep(0.05)
# time.sleep(0.05) in coroutine blocks ALL tasks - avoidasyncio.sleepis awaitable and non-blocking for other tasks.time.sleepfreezes entire event loop thread.- Same rule for sync HTTP, sync DB drivers in coroutines.
Related: Sync/Async Bridging - offload blocking
6. Timeout with wait_for
Bound waiting time on slow coroutines.
import asyncio
async def slow() -> str:
await asyncio.sleep(1)
return "ok"
async def main() -> None:
try:
await asyncio.wait_for(slow(), timeout=0.1)
except TimeoutError:
print("timed out")
asyncio.run(main())- Raises
TimeoutErroron expiry. - Cancels the underlying task on timeout.
- Prefer
asyncio.timeoutcontext manager (3.11+) in new code.
Related: Structured Concurrency - timeout patterns
Intermediate Examples
7. async with Resource Pattern
Async context managers release resources without blocking peers.
import asyncio
class AsyncResource:
async def __aenter__(self):
await asyncio.sleep(0)
return self
async def __aexit__(self, *args):
await asyncio.sleep(0)
async def main() -> None:
async with AsyncResource() as res:
print("using", res)
asyncio.run(main())__aenter__/__aexit__mirror syncwith.- Use for connections, locks, sessions.
- See dedicated page for streaming iterators.
Related: Async Context Managers & Iterators
8. Loop-Bound CPU Work Smell
CPU loops starve other coroutines.
import asyncio
async def bad_cpu() -> int:
total = 0
for i in range(500_000):
total += i
return total
# Fix: await asyncio.to_thread(cpu_intensive_fn)- Long sync sections block heartbeat and health checks.
- Offload to thread or process pool.
- Profile before optimizing - small loops are fine.
Related: Sync/Async Bridging - to_thread
9. FastAPI Lifespan Entry
Servers keep one long-lived loop - not repeated asyncio.run per request.
# Conceptual FastAPI 0.115+ pattern
# @asynccontextmanager async def lifespan(app):
# yield
# app = FastAPI(lifespan=lifespan)- Framework owns loop lifetime.
- Open shared
httpx.AsyncClientin lifespan startup. - Close resources on shutdown.
Related: Async HTTP with httpx/aiohttp - shared client
Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), 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+.