Asyncio Essentials
Concurrent IO with asyncio - tasks, timeouts, and structured concurrency basics. Results appear in the same fence: same-line # when short, multiline # blocks below the sample when not.
//
Busque em todas as páginas da documentação
Concurrent IO with asyncio - tasks, timeouts, and structured concurrency basics. Results appear in the same fence: same-line # when short, multiline # blocks below the sample when not.
Define coroutines with async def and wait on awaitables with await.
import asyncio
async def add(a: int, b: int) -> int:
await asyncio.sleep(0)
return a + b
asyncio.run(add(2, 3)) # 5Schedule a coroutine to run concurrently on the running loop.
import asyncio
async def main():
t = asyncio.create_task(asyncio.sleep(0, result=42))
return await t
asyncio.run(main()) # 42Run many coroutines; optionally collect exceptions as results.
import asyncio
async def ok(): return 1
async def bad(): raise ValueError("x")
async def main():
return await asyncio.gather(ok(), bad(), return_exceptions=True)
r = asyncio.run(main())
r[0], type(r[1]).__name__ # (1, 'ValueError')Cancel a coroutine that exceeds a deadline.
import asyncio
async def main():
try:
await asyncio.wait_for(asyncio.sleep(10), timeout=0.01)
return "ok"
except TimeoutError:
return "timeout"
asyncio.run(main()) # 'timeout'Producer/consumer coordination with an async queue.
import asyncio
async def main():
q: asyncio.Queue[str] = asyncio.Queue()
await q.put("job")
return await q.get()
asyncio.run(main()) # 'job'Async context managers acquire/release resources with awaitable enter/exit.
import asyncio
async def main():
lock = asyncio.Lock()
async with lock:
return lock.locked() # True inside
asyncio.run(main()) # TrueIterate async iterators (streams, cursors) with async for.
import asyncio
async def gen():
for i in range(3):
yield i
async def main():
return [x async for x in gen()]
asyncio.run(main()) # [0, 1, 2]Limit concurrency into shared resources.
import asyncio
async def main():
sem = asyncio.Semaphore(1)
async with sem:
held = True
return held
asyncio.run(main()) # TrueSignal between tasks with Event.
import asyncio
async def main():
ready = asyncio.Event()
ready.set()
await ready.wait()
return ready.is_set()
asyncio.run(main()) # TrueRun blocking functions in a thread so the loop stays responsive (3.9+).
import asyncio
async def main():
return await asyncio.to_thread(lambda: sum(range(1000)))
asyncio.run(main()) # 499500Structured concurrency with TaskGroup (3.11+) - cancel siblings on failure.
import asyncio
async def main():
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(asyncio.sleep(0, result=1))
t2 = tg.create_task(asyncio.sleep(0, result=2))
return t1.result() + t2.result()
asyncio.run(main()) # 3Protect a awaitable from cancellation carefully with shield.
import asyncio
async def main():
return await asyncio.shield(asyncio.sleep(0, result="done"))
asyncio.run(main()) # 'done'Start the event loop from sync code with asyncio.run.
import asyncio
async def main():
return "started"
asyncio.run(main()) # 'started'Process results as soon as each task finishes.
import asyncio
async def main():
tasks = [asyncio.sleep(0, result=i) for i in (2, 1, 3)]
return [await c for c in asyncio.as_completed(tasks)]
sorted(asyncio.run(main())) # [1, 2, 3]Async timeout context manager (3.11+) for blocks of awaits.
import asyncio
async def main():
try:
async with asyncio.timeout(0.01):
await asyncio.sleep(10)
return "ok"
except TimeoutError:
return "timeout"
asyncio.run(main()) # 'timeout'Stack versions: Python 3.14.0 (stable 3.14, maintenance 3.13) · uv 0.6+ · ruff 0.9+
Revisado por Chris St. John·Última atualização: 31 de jul. de 2026