Async/Await: Cooperative Concurrency

A web scraper that fetches 100 URLs spends 99% of its life waiting — DNS, TCP handshake, server response, body bytes. A single thread issuing requests.get in a loop is idle for almost every microsecond of its existence. Spin up 100 threads and you trade idleness for context-switch overhead, lock contention, and ~8 MB of stack per thread.

async/await solves the same problem with one thread and zero threads. A coroutine voluntarily pauses at each await; the event loop picks another ready coroutine and runs that until its next pause. Hundreds of concurrent network calls share one OS thread and a few kilobytes of state each. It is concurrency without parallelism — and for I/O-bound work, that is exactly what you want.

This lesson is the mental model, the syntax, the standard-library API, and the half-dozen mistakes that cause every async bug in production.

1. The Problem — Blocking I/O Wastes Wall Clock

python

import time
import requests

def fetch(url):
    return requests.get(url).status_code

urls = ["https://example.com"] * 10

start = time.perf_counter()
results = [fetch(u) for u in urls]
print(f"sync: {time.perf_counter() - start:.2f}s")     # ~2.5s — each call blocks

Ten requests, each ~250 ms of network, executed strictly one after the other. The CPU is idle the whole time. You could be doing the next nine requests during the wait for the first one — but a synchronous program has no way to express "wait, but let other work run".

Threads can express it, but they're expensive. Async does it on one thread.

2. The Mental Model — One Thread, Many Tasks

Imagine a single chef in a kitchen with ten ovens. Synchronous code puts a pie in oven 1 and stares at it for 30 minutes. Async code starts oven 1, walks to oven 2, starts that one, walks to oven 3, etc. When a timer rings the chef pulls that pie out and starts the next one. One chef, ten pies cooking concurrently.

The chef is the OS thread. The ovens are I/O operations (sockets, files, subprocesses). The kitchen timer is the event loop. The pie-walking is await.

Two non-negotiable rules:

1. The chef never sits down. If your coroutine does CPU work (a tight numeric loop, hashing 100 MB, parsing a 10 MB JSON), no other task can run. Async is for I/O.
2. The chef can only switch tasks at await. A coroutine without any await is just a slow synchronous function with extra syntax.

3. `async def` Defines a Coroutine

python

async def hello():
    return "hi"

print(hello())                       # <coroutine object hello at 0x...>
                                     # NOT "hi" — the body did not run

Calling an async def function does not execute it. It returns a coroutine object — a recipe for work that the event loop can drive. Two ways to actually run it:

python

import asyncio

asyncio.run(hello())                 # entry point — creates a loop, runs to completion

# or, from inside another coroutine:
async def caller():
    result = await hello()           # await drives the coroutine, returns its value
    print(result)                    # "hi"

asyncio.run(caller())

+ setup added so this can run · defines hello

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def hello(*_a, **_kw):
    print('-> hello() called')
    return _AutoMock('hello()')

asyncio.run() is the one and only way to start the event loop from synchronous code. Don't call it from inside another coroutine — you'd be starting a loop while one is already running. Use await instead.

4. The Canonical Demo — Sequential vs Concurrent

python

import asyncio
import time

async def fetch(name, delay):
    print(f"  start {name}")
    await asyncio.sleep(delay)                   # cooperative — yields to the loop
    print(f"  done  {name}")
    return name.upper()

# Sequential — await one at a time
async def sequential():
    t0 = time.perf_counter()
    a = await fetch("a", 1)
    b = await fetch("b", 1)
    c = await fetch("c", 1)
    print(f"sequential: {time.perf_counter() - t0:.2f}s")
    return [a, b, c]

# Concurrent — start all three, then wait
async def concurrent():
    t0 = time.perf_counter()
    results = await asyncio.gather(
        fetch("a", 1),
        fetch("b", 1),
        fetch("c", 1),
    )
    print(f"concurrent: {time.perf_counter() - t0:.2f}s")
    return results

asyncio.run(sequential())            # ~3.0s — one after the other
asyncio.run(concurrent())            # ~1.0s — all three overlap

asyncio.sleep is the async-aware sleep — it tells the event loop "wake me in N seconds; run something else meanwhile." time.sleep would block the entire loop for 3 seconds and produce the same wall time as the sync version. time.sleep in async code is the most common async bug.

5. `asyncio.gather` — Run Many, Wait For All

python

async def main():
    results = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )
    # results is a list, in the same order as the args
    for user in results:
        print(user)

asyncio.run(main())

+ setup added so this can run · defines asyncio, fetch_user

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

asyncio = _AutoMock('asyncio')
def fetch_user(*_a, **_kw):
    print('-> fetch_user() called')
    return _AutoMock('fetch_user()')

gather schedules all coroutines on the event loop, then waits for every one to finish. Results come back in the order the coroutines were passed — not the order they completed. If one raises, the exception propagates from gather (and by default the others keep running; pass return_exceptions=True to collect exceptions instead of raising).

gather is fine for fire-and-forget batches. For anything where one failure should cancel the rest — and that's most production code — use TaskGroup (Section 7).

6. `asyncio.create_task` — Background Work

await coro runs coro to completion before continuing. Sometimes you want to start work and continue immediately, awaiting the result later. asyncio.create_task schedules a coroutine on the loop and returns a Task handle.

python

async def main():
    task = asyncio.create_task(fetch("background", 2))
    print("kicked off the fetch; doing other work...")
    await asyncio.sleep(0.5)
    print("still working...")
    result = await task              # now block until the task finishes
    print("got:", result)

asyncio.run(main())

+ setup added so this can run · defines asyncio, fetch

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

asyncio = _AutoMock('asyncio')
def fetch(*_a, **_kw):
    print('-> fetch() called')
    return _AutoMock('fetch()')

A Task is a wrapped coroutine that's already running on the loop. gather and TaskGroup are built on top of it. Useful patterns:

python

# Fire many, await later
tasks = [asyncio.create_task(fetch(u)) for u in urls]
results = [await t for t in tasks]   # works, but TaskGroup is better

# Cancel a runaway task
task = asyncio.create_task(forever())
await asyncio.sleep(1)
task.cancel()                        # raises CancelledError inside the task
try:
    await task
except asyncio.CancelledError:
    print("clean shutdown")

7. `asyncio.TaskGroup` — Structured Concurrency (3.11+)

TaskGroup is the modern, recommended way to launch concurrent work. If any task fails, all sibling tasks are cancelled and the group raises an ExceptionGroup. Resources are cleaned up. No orphans.

python

import asyncio

async def fetch(name, delay):
    await asyncio.sleep(delay)
    if name == "b":
        raise ValueError("b is broken")
    return name.upper()

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch("a", 1))
        t2 = tg.create_task(fetch("b", 0.5))     # this raises
        t3 = tg.create_task(fetch("c", 2))       # auto-cancelled when b fails
    # block exits — all tasks done. If any failed, ExceptionGroup is raised here.
    print(t1.result(), t3.result())

try:
    asyncio.run(main())
except* ValueError as eg:                        # PEP 654 except* syntax
    for e in eg.exceptions:
        print("caught:", e)

+ setup added so this can run · defines tg

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

tg = _AutoMock('tg')

Three reasons TaskGroup beats raw gather:

Cancellation safety — one failure cancels its siblings; you don't leak tasks running in the background.
Scoped lifetime — the async with block guarantees every task finishes (or is cancelled) before code after the block runs.
ExceptionGroup — multiple simultaneous failures are bundled rather than swallowed.

On Python ≤ 3.10, fall back to gather(..., return_exceptions=True) and check for exceptions yourself.

8. Timeouts — `wait_for` and `asyncio.timeout`

A coroutine that never completes will keep your program alive forever. Set timeouts.

python

# Old API — wraps a single coroutine
try:
    result = await asyncio.wait_for(fetch_slow(), timeout=2.0)
except asyncio.TimeoutError:
    print("gave up after 2s")

# Modern API (3.11+) — works as a context manager around any code
try:
    async with asyncio.timeout(2.0):
        result = await fetch_slow()
        # any number of awaits inside; the whole block is bounded
except asyncio.TimeoutError:
    print("gave up after 2s")

asyncio.timeout composes with TaskGroup, multiple awaits, and async with blocks — wait_for only wraps a single coroutine. Use timeout on new code.

asyncio.TimeoutError is an alias for the built-in TimeoutError in 3.11+.

9. `async for` and Async Generators

Just like def + yield gives you a generator, async def + yield gives you an async generator — consumed with async for.

python

async def stream_pages(api):
    page = 1
    while True:
        items = await api.get_page(page)         # async I/O between yields
        if not items:
            return
        for item in items:
            yield item
        page += 1

async def main():
    async for item in stream_pages(api):
        print(item)

This is how paginated HTTP APIs, websockets, and server-sent events get consumed. Each yield produces one item; each await can pause the loop. The mental model from generators carries over directly — lazy, one-at-a-time, paused between yields, just awaitable.

10. `async with` — Async Context Managers

The async cousin of with. The __aenter__ and __aexit__ methods are coroutines, so opening and closing can do I/O.

python

import httpx

async def main():
    async with httpx.AsyncClient() as client:        # __aenter__ might do TCP setup
        r = await client.get("https://api.example.com/users")
        return r.json()
    # __aexit__ closes the connection pool, awaiting any pending cleanup

+ setup added so this can run · defines client

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

client = _AutoMock('client')

You'll meet async with on every HTTP client, every database connection pool, every async lock. Build your own with @contextlib.asynccontextmanager — see contextmanagers for the sync version; the async one is structurally identical.

11. Concurrent ≠ Parallel — The CPU Trap

Async runs on one thread. One thread on a multi-core machine uses one core. CPU-bound work in a coroutine blocks the event loop for the entire duration of the calculation.

python

async def hash_a_lot(data):
    # BAD — synchronous CPU work; loop is frozen for the duration
    return hashlib.sha256(data).hexdigest()

async def main():
    await asyncio.gather(*[hash_a_lot(blob) for blob in big_blobs])
    # NO speedup — these run strictly one after another

+ setup added so this can run · defines asyncio, hashlib, big_blobs

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

asyncio = _AutoMock('asyncio')
hashlib = _AutoMock('hashlib')
big_blobs = ["alpha", "beta", "gamma"]

Async is concurrency for I/O-bound work. For CPU-bound parallelism, use multiprocessing — it bypasses the GIL with separate processes, and the concurrent.futures.ProcessPoolExecutor API composes cleanly with async (next section).

Workload	Tool
Many I/O calls, modern libs	`asyncio`
Few I/O calls, blocking libs	`ThreadPoolExecutor`
CPU-bound	`ProcessPoolExecutor`

12. Mixing Sync Into Async — `run_in_executor`

Sometimes the library you need only has a blocking API (requests, the standard sqlite3, a slow file parser, a hash). Don't call it directly from a coroutine — you'd block the loop. Push it off to a thread or process pool:

python

import asyncio
import requests
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

async def fetch_with_blocking_lib(url):
    loop = asyncio.get_running_loop()
    # None = use the default ThreadPoolExecutor
    return await loop.run_in_executor(None, requests.get, url)

# CPU-bound — push to a process pool
def heavy_hash(data):
    return hashlib.sha256(data * 1000).hexdigest()

async def hash_many(blobs):
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        results = await asyncio.gather(
            *[loop.run_in_executor(pool, heavy_hash, b) for b in blobs]
        )
    return results

+ setup added so this can run · defines hashlib

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

hashlib = _AutoMock('hashlib')

run_in_executor returns a future that integrates with the event loop — you await it like any coroutine. The blocking call runs on a worker thread (or process); the loop stays responsive.

The 3.9+ shortcut: await asyncio.to_thread(blocking_fn, *args) — same idea, less ceremony.

13. The Real-World Ecosystem

You will almost never write async code that talks directly to a socket. The async ecosystem gives you high-level libraries that already speak the protocol.

Need	Library
HTTP client	`httpx.AsyncClient`, `aiohttp`
HTTP server	`FastAPI`, `Starlette`, `aiohttp`
PostgreSQL	`asyncpg`, `psycopg` (3.x)
Redis	`redis.asyncio`
File I/O	`aiofiles` (thin wrapper over threads)
Subprocesses	`asyncio.create_subprocess_exec`
Testing	`pytest-asyncio`, `anyio`

python

import httpx, asyncio

async def fetch_json(url):
    async with httpx.AsyncClient(timeout=5.0) as client:
        r = await client.get(url)
        r.raise_for_status()
        return r.json()

async def main():
    urls = ["https://api.example.com/u/1", "https://api.example.com/u/2"]
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_json(u)) for u in urls]
    for t in tasks:
        print(t.result())

asyncio.run(main())

+ setup added so this can run · defines client, tg

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

client = _AutoMock('client')
tg = _AutoMock('tg')

That snippet is production-shaped: connection pooling via the client context manager, structured concurrency via TaskGroup, timeouts on the client, and clean propagation of HTTP errors.

Common Mistakes

1. Forgetting await

python

async def main():
    fetch("a", 1)                    # creates a coroutine object, never runs it
                                     # RuntimeWarning: coroutine 'fetch' was never awaited

+ setup added so this can run · defines fetch

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def fetch(*_a, **_kw):
    print('-> fetch() called')
    return _AutoMock('fetch()')

A bare coroutine call produces a coroutine object and discards it. The body never runs. The warning is easy to miss in noisy logs. Linters (ruff, pylint) catch this — turn the rule on.

2. Calling time.sleep or requests.get from a coroutine

Any synchronous blocking call freezes the event loop. Every other task waits. Symptoms: "my async server has the same throughput as the sync one." Use asyncio.sleep, httpx.AsyncClient, asyncpg — async-native replacements — or push the blocking call through asyncio.to_thread.

3. asyncio.run() inside a running loop

python

async def main():
    asyncio.run(other())             # RuntimeError: asyncio.run() cannot be called
                                     # from a running event loop

+ setup added so this can run · defines asyncio, other

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

asyncio = _AutoMock('asyncio')
def other(*_a, **_kw):
    print('-> other() called')
    return _AutoMock('other()')

asyncio.run creates a new loop. You can't nest loops. From inside a coroutine, use await other(). asyncio.run belongs at the top of if __name__ == "__main__": and nowhere else.

4. Unbounded fan-out

python

async def fetch_all(urls):
    return await asyncio.gather(*[fetch(u) for u in urls])
    # 10,000 URLs → 10,000 simultaneous connections → kernel sockets exhausted,
    # remote rate-limits trigger, OOM

+ setup added so this can run · defines asyncio, fetch

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

asyncio = _AutoMock('asyncio')
def fetch(*_a, **_kw):
    print('-> fetch() called')
    return _AutoMock('fetch()')

Cap concurrency with a Semaphore:

python

async def fetch_all(urls, concurrency=20):
    sem = asyncio.Semaphore(concurrency)
    async def bounded(u):
        async with sem:
            return await fetch(u)
    return await asyncio.gather(*[bounded(u) for u in urls])

+ setup added so this can run · defines asyncio, fetch

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

asyncio = _AutoMock('asyncio')
def fetch(*_a, **_kw):
    print('-> fetch() called')
    return _AutoMock('fetch()')

5. Leaking tasks on shutdown

asyncio.create_task returns a Task that runs in the background. If you never await it and the program exits, you get a Task was destroyed but it is pending! warning. Worse, in long-running services, orphaned tasks pile up. Use TaskGroup; or keep references and await them; or wrap them so you cancel and await on shutdown.

6. Treating async as a speedup for CPU work

asyncio does not run your code in parallel. Same thread, same GIL, same one core. For CPU-bound parallelism, multiprocessing.

🎯 Your Turn — Bounded Concurrent Fetcher

Write async def fetch_all(urls, concurrency=10) that fetches a list of URLs concurrently, with no more than concurrency requests in flight at once. Return a list of (url, status_code) tuples in the same order as the input URLs.

Constraints:

Use asyncio.Semaphore to cap in-flight requests.
Use asyncio.gather (or TaskGroup) to launch the work.
Use httpx.AsyncClient if available; otherwise simulate the request with await asyncio.sleep(...) and a fake status code so the structure is testable without network.
If any single fetch raises, return (url, None) for that URL — do not let one failure kill the batch.

python

import asyncio
# import httpx  # uncomment if installed

async def fetch_all(urls, concurrency=10):
    # TODO 1: create a Semaphore(concurrency)
    # TODO 2: define an inner coroutine `bounded_fetch(url)` that:
    #          - acquires the semaphore via `async with sem:`
    #          - performs the request (httpx) or stand-in (asyncio.sleep)
    #          - returns (url, status_code), or (url, None) on exception
    # TODO 3: await asyncio.gather(*[bounded_fetch(u) for u in urls])
    # TODO 4: return the results list
    ...

# Demo
urls = [f"https://example.com/{i}" for i in range(50)]
print(asyncio.run(fetch_all(urls, concurrency=5)))

Hint 1 — Semaphore as an async context manager

asyncio.Semaphore supports async with: async with sem: acquires on entry and releases on exit (even if an exception is raised inside). That's the safe pattern — never call .acquire()/.release() manually unless you have to.

Hint 2 — Per-task error handling

If you want one bad URL not to break the batch, wrap the network call in try/except inside bounded_fetch and return (url, None) from the except branch. Alternatively pass return_exceptions=True to gather and translate exceptions to None afterwards.

Show full solution

python

import asyncio
import random

try:
    import httpx
    HAS_HTTPX = True
except ImportError:
    HAS_HTTPX = False


async def fetch_all(urls, concurrency=10):
    """Fetch URLs concurrently with at most `concurrency` in flight.

    Returns list of (url, status_code) tuples in input order.
    Failed fetches return (url, None).
    """
    sem = asyncio.Semaphore(concurrency)

    async def bounded_fetch(client, url):
        async with sem:
            try:
                if HAS_HTTPX and client is not None:
                    r = await client.get(url, timeout=5.0)
                    return (url, r.status_code)
                else:
                    # Stand-in: simulate variable latency and the odd failure
                    await asyncio.sleep(random.uniform(0.05, 0.2))
                    if random.random() < 0.05:
                        raise RuntimeError("simulated network blip")
                    return (url, 200)
            except Exception:
                return (url, None)

    if HAS_HTTPX:
        async with httpx.AsyncClient() as client:
            return await asyncio.gather(
                *[bounded_fetch(client, u) for u in urls]
            )
    else:
        return await asyncio.gather(
            *[bounded_fetch(None, u) for u in urls]
        )


# Demo
async def demo():
    urls = [f"https://example.com/{i}" for i in range(50)]
    results = await fetch_all(urls, concurrency=5)
    ok = sum(1 for _, s in results if s == 200)
    print(f"got {len(results)} results; {ok} ok, {len(results)-ok} failed")

asyncio.run(demo())

The shape of this solution is the production template for "fan out N network calls, cap the fan-out, don't let one failure kill the batch":

One shared AsyncClient — reuses the connection pool across all tasks. Building a new client per request is the most common perf regression in async HTTP code.
async with sem — acquires before the network call, releases on the way out, even if the call raises. No manual bookkeeping.
Per-task try/except — converts failures into a sentinel value so gather doesn't bail out on the first error. If you'd rather see exceptions, drop the try/except and pass return_exceptions=True to gather.
Order preserved — gather returns results in the order coroutines were passed, regardless of completion order. That's why we can pair them back up to the input URLs by index.

For very large batches (100k+ URLs), upgrade to TaskGroup so a KeyboardInterrupt or a fatal asyncio.timeout cancels every in-flight request cleanly. The semaphore stays the same.

What You Learned

async def defines a coroutine function; calling it returns a coroutine object. The body runs only under await or asyncio.run.
asyncio.run(coro) is the synchronous entry point. Inside a coroutine, use await.
asyncio.sleep yields to the event loop; time.sleep blocks it. Same goes for requests.get vs httpx.AsyncClient.
asyncio.gather runs many coroutines concurrently, returns results in input order.
asyncio.create_task schedules a coroutine to run in the background; await the returned Task for its result.
asyncio.TaskGroup (3.11+) is the modern, structured-concurrency replacement for gather — auto-cancels siblings on failure, surfaces ExceptionGroup.
Timeouts: asyncio.wait_for (single coro) and asyncio.timeout() (any block) — pick the latter.
async for consumes async generators; async with wraps async context managers — connection pools, async locks, async clients.
Concurrent != parallel. Async is for I/O-bound work on one thread. CPU-bound → multiprocessing.
asyncio.to_thread(fn, ...) (or loop.run_in_executor) pushes blocking calls off the loop.
Cap fan-out with asyncio.Semaphore. Unbounded gather exhausts sockets, rate-limits remote APIs, and OOMs your process.
The ecosystem: httpx, aiohttp, FastAPI, asyncpg, redis.asyncio — async-native libraries are now the default.

Next: Multiprocessing & Threading — when one thread isn't enough, when the GIL gets in your way, and the decision matrix for picking the right concurrency model.

1. The Problem — Blocking I/O Wastes Wall Clock

2. The Mental Model — One Thread, Many Tasks

3. async def Defines a Coroutine

4. The Canonical Demo — Sequential vs Concurrent

5. asyncio.gather — Run Many, Wait For All

6. asyncio.create_task — Background Work

7. asyncio.TaskGroup — Structured Concurrency (3.11+)

8. Timeouts — wait_for and asyncio.timeout

9. async for and Async Generators

10. async with — Async Context Managers

11. Concurrent ≠ Parallel — The CPU Trap

12. Mixing Sync Into Async — run_in_executor

13. The Real-World Ecosystem

Common Mistakes

🎯 Your Turn — Bounded Concurrent Fetcher

What You Learned

3. `async def` Defines a Coroutine

5. `asyncio.gather` — Run Many, Wait For All

6. `asyncio.create_task` — Background Work

7. `asyncio.TaskGroup` — Structured Concurrency (3.11+)

8. Timeouts — `wait_for` and `asyncio.timeout`

9. `async for` and Async Generators

10. `async with` — Async Context Managers

12. Mixing Sync Into Async — `run_in_executor`