The essentials

Quick reference

One focused task per row. Jump to the related section for complete, working examples.

UseSyntaxExamples
Define a coroutineasync def fetch(): return await client.get('/')View examples
Run an entry coroutineasyncio.run(main())View examples
Await completionresult = await operation()View examples
Schedule a tasktask = asyncio.create_task(operation())View examples
Group child tasksasync with asyncio.TaskGroup() as group: group.create_task(operation())View examples
Collect ordered resultsresults = await asyncio.gather(*coroutines)View examples
Bound a block by timeasync with asyncio.timeout(5): result = await operation()View examples
Request cancellationtask.cancel()View examples
Yield to the loopawait asyncio.sleep(0)View examples
Create a bounded queuequeue = asyncio.Queue(maxsize=100)View examples
Bound concurrent accessasync with asyncio.Semaphore(10): await operation()View examples
Run blocking I/O in a threadresult = await asyncio.to_thread(blocking_function, argument)View examples

asyncio coordinates many waiting operations on one event loop. A coroutine does nothing until awaited or scheduled; structured concurrency keeps child lifetimes tied to a scope. Preserve cancellation, bound concurrency, and move blocking work away from the loop thread.

Step by step

Detailed examples

01

Run coroutine objects exactly once

Calling async def creates a coroutine object but does not execute its body. Await it inside async code or use asyncio.run at a synchronous program boundary. Do not call asyncio.run from a thread that already has a running event loop.

Small async entry point
import asyncio

async def greet(name):
    await asyncio.sleep(0)
    return f'hello {name}'

async def main():
    print(await greet('Ada'))

asyncio.run(main())
Output
hello Ada
Back to quick reference ↑
02

Prefer scoped concurrency

create_task starts independent progress but you must retain and await its handle. TaskGroup waits for all children on exit and cancels siblings after a non-cancellation failure, then raises an exception group. gather preserves input order but has different failure and cancellation semantics.

Collect results from a task group
import asyncio

async def square(value):
    await asyncio.sleep(0)
    return value * value

async def main():
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(square(n)) for n in range(3)]
    print([task.result() for task in tasks])

asyncio.run(main())
Output
[0, 1, 4]
Back to quick reference ↑
03

Treat cancellation as control flow

Cancellation raises CancelledError at an await point. Use try/finally for cleanup and normally re-raise cancellation after any necessary work. asyncio.timeout bounds a region; swallowing CancelledError can break TaskGroup and timeout behavior.

Handle an expected timeout
import asyncio

async def main():
    try:
        async with asyncio.timeout(0.01):
            await asyncio.sleep(1)
    except TimeoutError:
        print('timed out')

asyncio.run(main())
Output
timed out
Back to quick reference ↑
04

Bound producers and shared capacity

Queue provides async put/get and optional capacity; call task_done once per completed get when using join. Semaphore limits simultaneous entries but does not guarantee fairness. Locks protect coroutine-level invariants, not code running in other threads or processes.

Bounded producer and consumer
import asyncio

async def main():
    queue = asyncio.Queue(maxsize=2)
    for item in ['a', 'b']:
        await queue.put(item)
    while not queue.empty():
        print(await queue.get())
        queue.task_done()
    await queue.join()

asyncio.run(main())
Output
a
b
Back to quick reference ↑
05

Keep blocking calls off the event loop

A regular blocking function freezes every task on the loop thread. asyncio.to_thread is primarily for I/O-bound blocking APIs and propagates context variables. CPU-bound work usually needs a process pool or native code that releases the GIL; cancellation of the await does not necessarily stop underlying thread work.

Delegate a blocking function
import asyncio

def blocking_add(left, right):
    return left + right

async def main():
    result = await asyncio.to_thread(blocking_add, 2, 3)
    print(result)

asyncio.run(main())
Output
5
Back to quick reference ↑

Local code tester

Run structured async work

Create scoped tasks, bound their wait time, and collect results without blocking the loop.

Runs in your browser
Output
Press Run to load Python locally.

Sources and further reading

References

Authoritative documentation used to verify and expand this cheat sheet.

  1. Python Software Foundationasyncio — Asynchronous I/Odocs.python.org
  2. Python Software FoundationCoroutines and Tasksdocs.python.org
  3. Python Software FoundationQueuesdocs.python.org
  4. Python Software FoundationSynchronization Primitivesdocs.python.org

Help us improve

Found a typo or missing example?

Tell us what would make this cheat sheet clearer, more complete, or more useful.

Share feedback