asyncio
asyncio is the standard-library framework that provides the event loop, Task/Future scheduling, and coordination primitives that make async/await actually run and cooperate.
asyncio is the standard-library package that implements Python's async I/O framework: it provides the event loop that drives coroutines, the Task and Future objects used to schedule and track concurrent work, high-level coordination functions like gather() and wait(), timeout and cancellation support, and asyncio-flavored synchronization primitives (locks, events, queues) analogous to those in threading.
Event Loop
The event loop is the central engine asyncio provides: it runs coroutines, dispatches ready callbacks, handles I/O readiness notifications, and manages timers. Modern asyncio code almost never creates or manages the loop directly -- asyncio.run(coro) is the standard entry point: it creates a new event loop, runs the given coroutine to completion, and cleans up the loop afterward. It should be called once, from your program's synchronous top level (not from inside another coroutine, and generally not called more than once per program).
import asyncio
async def main() -> None:
print("running inside the event loop")
await asyncio.sleep(0.1)
print("done")
asyncio.run(main()) # creates the loop, runs main() to completion, closes the loopTasks
A Task wraps a coroutine and schedules it to run on the event loop concurrently with other code, rather than waiting for it to be awaited directly. asyncio.create_task(coro) is the standard way to start a coroutine running in the background: it returns immediately with a Task object, and the wrapped coroutine begins making progress at the next opportunity the event loop gets, even before you await the Task.
create_task() schedules the coroutine to start running immediately (concurrently), unlike a bare await, which would run it to completion before moving to the next line.
A common gotcha: creating a Task and never awaiting or otherwise keeping a reference to it risks the task being garbage-collected mid-flight in some circumstances; asyncio documentation recommends keeping a reference to tasks you create (e.g. in a set) until they complete, unless using a TaskGroup (3.11+), which manages this automatically.
import asyncio
async def background_work(label: str, seconds: float) -> str:
await asyncio.sleep(seconds)
return f"{label} done"
async def main() -> None:
task1 = asyncio.create_task(background_work("A", 2))
task2 = asyncio.create_task(background_work("B", 1))
# Both tasks are now running concurrently in the background
print("tasks started, doing other work...")
result1 = await task1
result2 = await task2
print(result1, result2)
asyncio.run(main()) # total time ~2s, not 3sFutures
A Future is a lower-level awaitable representing a result that may not be available yet -- essentially a placeholder for a value (or exception) that will be set at some point. Task is actually a subclass of Future specialized to wrap and drive a coroutine. Most application code interacts with Tasks (and coroutines) directly and rarely needs to construct a bare Future by hand -- Futures are more commonly encountered as the return type of certain lower-level APIs (e.g. bridging callback-based code into asyncio) or under the hood.
create_task, gather, and wait
asyncio.create_task(coro) schedules one coroutine to run concurrently, returning a Task. asyncio.gather(*awaitables) runs several awaitables concurrently and returns their results as a list, in the same order they were passed, once all complete (it raises the first exception encountered by default, though return_exceptions=True collects exceptions as results instead of raising). asyncio.wait(tasks) is a lower-level alternative that returns (done, pending) sets of tasks, useful when you want more control -- e.g. proceeding as soon as the first task finishes (FIRST_COMPLETED) rather than waiting for all of them.
Since Python 3.11, asyncio.TaskGroup is the recommended structured-concurrency alternative to gather()/wait() for many cases: it manages a group of tasks as a unit, automatically cancels sibling tasks if one fails, and surfaces multiple failures together as an ExceptionGroup (see the Async Context Managers & Iterators and general Exceptions topics for related mechanics).
import asyncio
async def maybe_fail(n: int) -> int:
await asyncio.sleep(0.1)
if n == 2:
raise ValueError(f"bad value: {n}")
return n * 10
async def main() -> None:
# gather with return_exceptions=True collects errors instead of raising
results = await asyncio.gather(
maybe_fail(1), maybe_fail(2), maybe_fail(3),
return_exceptions=True,
)
print(results) # [10, ValueError('bad value: 2'), 30]
asyncio.run(main())import asyncio
async def fetch(n: int) -> int:
await asyncio.sleep(0.1)
return n * 2
async def main() -> None:
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(fetch(1))
t2 = tg.create_task(fetch(2))
print(t1.result(), t2.result()) # 2 4 -- both finished when the block exits
asyncio.run(main())Timeout
asyncio.timeout(seconds) (3.11+, used as an async with context manager) or the older asyncio.wait_for(awaitable, timeout) let you bound how long you're willing to wait for an operation. If the timeout elapses first, the awaited operation is cancelled and a TimeoutError is raised.
import asyncio
async def slow_operation() -> str:
await asyncio.sleep(5)
return "finished"
async def main() -> None:
try:
async with asyncio.timeout(1):
result = await slow_operation()
print(result)
except TimeoutError:
print("operation timed out after 1 second")
asyncio.run(main()) # prints "operation timed out after 1 second"Cancellation
Any Task can be cancelled via task.cancel(), which arranges for a CancelledError to be raised inside the coroutine at its next suspension point (the next await). Well-behaved coroutines let CancelledError propagate (optionally doing brief cleanup in a finally block) rather than swallowing it, since suppressing cancellation silently can leave the program in an inconsistent state or hang shutdown.
import asyncio
async def long_task() -> None:
try:
print("starting long task")
await asyncio.sleep(10)
print("this never prints if cancelled")
except asyncio.CancelledError:
print("long task was cancelled, cleaning up")
raise # re-raise so the cancellation actually takes effect
async def main() -> None:
task = asyncio.create_task(long_task())
await asyncio.sleep(0.5)
task.cancel()
try:
await task
except asyncio.CancelledError:
print("main observed the cancellation")
asyncio.run(main())Synchronization Primitives
asyncio provides coordination primitives modeled on threading's but designed to be awaited: asyncio.Lock (mutual exclusion between coroutines), asyncio.Event (a flag that coroutines can wait on until it's set), asyncio.Condition, asyncio.Semaphore (limit concurrent access to N at a time), and asyncio.Queue (an async-aware producer/consumer queue). These are asyncio-specific types -- do not mix them with threading.Lock or multiprocessing.Lock, which block the whole OS thread rather than yielding to the event loop, and would stall every other coroutine.
Only 2 workers run concurrently at any moment even though 5 are scheduled, because the Semaphore caps concurrent access.
import asyncio
async def limited_worker(sem: asyncio.Semaphore, n: int) -> None:
async with sem: # at most 2 workers run their "critical" section at once
print(f"worker {n} started")
await asyncio.sleep(1)
print(f"worker {n} finished")
async def main() -> None:
sem = asyncio.Semaphore(2)
await asyncio.gather(*(limited_worker(sem, i) for i in range(5)))
asyncio.run(main())Practical Usage
A typical asyncio program: define coroutines for each unit of I/O-bound work, use TaskGroup (3.11+) or gather() to run independent operations concurrently, wrap operations that might hang with asyncio.timeout(), and use asyncio.Semaphore to bound concurrency against external services with rate limits (e.g. at most 10 simultaneous HTTP requests) rather than firing thousands of requests unbounded.
Edge Cases and Pitfalls
Calling a blocking, synchronous function (file I/O without an async wrapper, a CPU-heavy computation, a classic blocking HTTP call) from inside a coroutine stalls the entire event loop for every task -- use asyncio.to_thread(fn, *args) to run it in a separate thread without blocking the loop. Creating a Task and letting it go out of scope without ever awaiting or storing it risks losing track of exceptions it raises (they may surface only as an 'exception was never retrieved' warning) or the task itself being cancelled by garbage collection.
Complete Example
import asyncio
async def fetch_resource(name: str, delay: float, sem: asyncio.Semaphore) -> str:
async with sem:
await asyncio.sleep(delay)
return f"{name}: ok"
async def main() -> None:
sem = asyncio.Semaphore(3) # at most 3 concurrent "requests"
resources = [("res-%d" % i, 0.5) for i in range(6)]
try:
async with asyncio.timeout(2):
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(fetch_resource(n, d, sem)) for n, d in resources]
except TimeoutError:
print("overall operation timed out")
return
for t in tasks:
print(t.result())
asyncio.run(main())Mental Model
Think of the event loop as a project manager, Tasks as employees assigned to run in parallel on their own paperwork, gather/TaskGroup as 'wait until this whole batch of employees is done', timeout as a deadline that cuts work short if it drags on, cancel as pulling an employee off a task early, and the synchronization primitives (Lock, Semaphore) as shared-resource rules ('only 3 people can use the printer at once') that keep the employees from tripping over each other.
Quick Reference
asyncio.run(coro) - entry point: create loop, run to completion
asyncio.create_task(coro) - schedule a coroutine to run concurrently
asyncio.gather(*aws, return_exceptions=False) - run many, collect results/errors
asyncio.wait(tasks, return_when=...) - lower-level: returns (done, pending)
asyncio.TaskGroup() (3.11+) - structured concurrency, auto-cancel on error
asyncio.timeout(seconds) (3.11+) - bound how long to wait
asyncio.wait_for(aw, timeout) - older per-awaitable timeout
task.cancel() / CancelledError - cooperative cancellation
asyncio.Lock/Event/Semaphore/Queue - async-aware coordination primitivesKey Takeaways
asyncio.run() is the standard entry point: it creates the event loop, runs your top-level coroutine to completion, and tears the loop down.
create_task()/TaskGroup schedule coroutines to run concurrently in the background; a bare await runs one coroutine to completion before moving on.
gather() runs multiple awaitables concurrently and collects results (or exceptions, with return_exceptions=True); TaskGroup (3.11+) is the modern structured-concurrency alternative with automatic sibling cancellation on failure.
asyncio.timeout()/wait_for() bound how long an operation may run; task.cancel() cooperatively cancels a running task by raising CancelledError at its next await.
Use asyncio's own Lock/Event/Semaphore/Queue, never threading's or multiprocessing's, inside coroutine code -- those block the whole thread instead of yielding to the loop.
Offload blocking synchronous calls with asyncio.to_thread() so they don't stall the entire event loop.
async def main(): asyncio.create_task(fire_and_forget()) # no reference kept await asyncio.sleep(5) # the task may be garbage-collected mid-flight, or its # exceptions silently lost async def main(): task = asyncio.create_task(fire_and_forget()) background_tasks.add(task) # keep a reference until done task.add_done_callback(background_tasks.discard) await asyncio.sleep(5)
A Task with no retained reference can be garbage-collected before completing, and any exception it raises may go unnoticed; keep a reference (or use a TaskGroup) until the task finishes.
import time async def handler(): time.sleep(3) # blocks the ENTIRE event loop, stalling every other coroutine return "done" import asyncio async def handler(): await asyncio.to_thread(time.sleep, 3) # runs in a separate thread return "done"
A blocking call inside a coroutine freezes the whole event loop; asyncio.to_thread() offloads it to a thread pool so other coroutines keep making progress.
async def worker(): try: await asyncio.sleep(10) except asyncio.CancelledError: pass # swallowed -- cancellation silently ignored async def worker(): try: await asyncio.sleep(10) except asyncio.CancelledError: print("cleaning up") raise # re-raise so cancellation actually propagates
Swallowing CancelledError without re-raising prevents the cancellation from actually taking effect, which can hang shutdown or leave the task appearing to run forever.
What does asyncio.run() do, and where should it be called from?
It creates a new event loop, runs the given coroutine to completion, and cleans up the loop afterward; it should be called once, from synchronous top-level code, not from inside another coroutine.
What is the difference between awaiting a coroutine directly and wrapping it with asyncio.create_task() first?
Awaiting a coroutine directly runs it to completion before the calling code continues; create_task() schedules it to start running concurrently in the background immediately, returning a Task you can await later while other code runs in between.
What is the relationship between Task and Future?
Task is a subclass of Future specialized to wrap and drive a coroutine; Future is the more general lower-level awaitable representing a result (or exception) that will be available at some point.
How does asyncio.gather(..., return_exceptions=True) differ from its default behavior?
By default, gather() raises the first exception encountered by any awaitable; with return_exceptions=True, exceptions are captured and returned in the results list alongside successful results instead of being raised.
What happens when you call task.cancel() on a running Task?
A CancelledError is raised inside the coroutine at its next suspension point (the next await); the coroutine should let this exception propagate (optionally doing brief cleanup) rather than swallowing it, so the cancellation actually takes effect.
Why can't you use threading.Lock inside asyncio coroutine code?
threading.Lock blocks the entire OS thread while waiting to acquire it, which would stall the whole event loop (and every other coroutine on it); asyncio.Lock is designed to yield control back to the event loop while waiting instead.
How do you safely run a blocking, synchronous function from within async code without stalling the event loop?
Offload it with asyncio.to_thread(fn, *args), which runs the blocking call in a separate thread pool thread and awaits its completion without blocking the event loop itself.
Write code using TaskGroup to run three coroutines concurrently and collect their results.
import asyncio async def compute(n: int) -> int: await asyncio.sleep(0.2) return n * n async def main() -> list[int]: async with asyncio.TaskGroup() as tg: tasks = [tg.create_task(compute(n)) for n in (2, 3, 4)] return [t.result() for t in tasks] print(asyncio.run(main())) # [4, 9, 16]
TaskGroup runs all three compute() coroutines concurrently; the async with block waits until all tasks complete (or cancels the rest if one fails) before exiting, after which each task's .result() is available.
What happens when this code runs?
It raises ValueError("bad") -- by default, asyncio.gather() propagates the first exception encountered among its awaitables rather than returning it as part of the results list; to collect it alongside successful results instead, gather() would need return_exceptions=True.
Write code that limits concurrent access to a resource to at most 2 coroutines at a time using a Semaphore, and demonstrate a timeout with asyncio.timeout().
import asyncio async def access_resource(sem: asyncio.Semaphore, n: int) -> None: async with sem: print(f"{n}: accessing") await asyncio.sleep(1) print(f"{n}: released") async def main() -> None: sem = asyncio.Semaphore(2) try: async with asyncio.timeout(1.5): await asyncio.gather(*(access_resource(sem, i) for i in range(4))) except TimeoutError: print("timed out before all workers finished") asyncio.run(main())
The Semaphore(2) ensures only 2 of the 4 access_resource coroutines run their critical section concurrently; the other two wait their turn. Because 4 workers must run 2-at-a-time for 1 second each (about 2 seconds total) but the timeout is only 1.5 seconds, the timeout block prints 'timed out before all workers finished'.
import asyncio
async def flaky(n):
if n == 2:
raise ValueError("bad")
return n
async def main():
results = await asyncio.gather(flaky(1), flaky(2), flaky(3))
print(results)
asyncio.run(main())