Three concurrency models — and asyncio is the modern default for I/O
Python has three: asyncio (single-threaded cooperative concurrency), threading (multiple threads, the GIL limits CPU parallelism), multiprocessing (multiple processes, true parallelism). For I/O-bound work — the network, the filesystem, anywhere you'd otherwise wait — asyncio is the modern default. cwkPippa is asyncio top to bottom.
async def and await
async def f(): defines a coroutine. Calling it doesn't run the body — it returns a coroutine object. await coro runs the coroutine to its next suspension or completion. The await keyword is the magic: it tells the event loop "you can run something else while we're waiting here."
asyncio.run — the entry point
asyncio.run(main()) creates an event loop, runs the coroutine until it returns, then shuts the loop down. This is how you start an async program from a regular Python script. From inside an async function, you don't call asyncio.run — you just await.
Tasks — running coroutines concurrently
await coro waits for ONE coroutine. To run several concurrently, wrap each in a Task with asyncio.create_task(coro). await asyncio.gather(*tasks) waits for all of them. asyncio.TaskGroup (3.11+) is the modern shape — block-scoped task management with proper exception handling.
Principle: "Async all the way" is a real constraint. You can't await inside a regular def. You can't call an async function without await (you'll get a coroutine object back, not a result). Once you decide a function is async, all its callers up the chain become async too.
Every created task also needs an owner. Losing the reference can lose its exception and obscure when it must finish. TaskGroup matters because every child completes or is cancelled before the scope exits: structured concurrency is a lifetime guarantee, not merely shorter syntax.
Code
Hello async world·python
import asyncio
async def hello():
print("hi")
await asyncio.sleep(0.5) # 'sleep' yields control to the loop
print("world")
# From a regular script — start the event loop
asyncio.run(hello())
# hi
# (half-second pause)
# world
Tasks vs sequential — concurrency wins·python
import asyncio
import time
async def fetch(name, delay):
print(f"{name}: starting")
await asyncio.sleep(delay)
print(f"{name}: done")
return name
async def sequential():
start = time.perf_counter()
a = await fetch("a", 1)
b = await fetch("b", 1)
c = await fetch("c", 1)
print(f"sequential: {time.perf_counter() - start:.2f}s")
async def concurrent():
start = time.perf_counter()
results = await asyncio.gather(
fetch("a", 1),
fetch("b", 1),
fetch("c", 1),
)
print(f"concurrent: {time.perf_counter() - start:.2f}s")
asyncio.run(sequential()) # ~3 seconds
asyncio.run(concurrent()) # ~1 second — they ran in parallel
create_task — fire and remember·python
import asyncio
async def background_work():
await asyncio.sleep(0.5)
print("background done")
async def main():
task = asyncio.create_task(background_work())
print("task scheduled")
# Do other work while it runs
await asyncio.sleep(0.1)
print("main is doing other things")
await task # wait for the background to finish
print("all done")
asyncio.run(main())
TaskGroup — modern structured concurrency (3.11+)·python
import asyncio
async def fetch(name, delay):
await asyncio.sleep(delay)
return f"{name}-result"
async def main():
async with asyncio.TaskGroup() as tg:
a = tg.create_task(fetch("a", 0.5))
b = tg.create_task(fetch("b", 0.3))
c = tg.create_task(fetch("c", 0.4))
# All tasks complete by here, exceptions raised as ExceptionGroup
print(a.result(), b.result(), c.result())
asyncio.run(main())
# Cleaner than gather + try/except for exception handling
Forgetting await — the most common bug·python
import asyncio
async def add(a, b):
return a + b
async def main():
# WRONG — no await, you get the coroutine, not the result
result = add(2, 3)
print(type(result)) # <class 'coroutine'>
print(result) # <coroutine object ...>
# RIGHT
result = await add(2, 3)
print(result) # 5
asyncio.run(main())
# Python warns about un-awaited coroutines on shutdown
Write async def fetch(url: str, delay: float) -> str that awaits a sleep(delay) then returns f"got {url}". Then write async def main() that runs three concurrent fetches with different delays using asyncio.gather. Time it with time.perf_counter. Then rewrite using asyncio.TaskGroup. Verify both versions produce the same output and same elapsed time.
Progress
Progress is local-only — sign in to sync across devices.