Async/Await: Cooperative Concurrency

A web scraper that fetches 100 URLs spends almost all of its time waiting: for DNS, for the TCP handshake, for the server response, and for the body bytes. A single thread calling requests.get in a loop is idle almost the whole time. If you start 100 threads instead, the waiting is shared, but you pay for context switches, lock contention, and memory for each thread's stack (often reserved as ~8 MB per thread).

async/await solves the same problem with one thread. A coroutine pauses by itself at each await. The event loop then picks another coroutine that is ready and runs that one until its next pause. Hundreds of concurrent network calls share one OS thread, and each one needs only a few kilobytes of state. This is concurrency without parallelism, and for I/O-bound work, that is what you want.

This lesson covers the mental model, the syntax, the standard-library API, and the six mistakes that cause most async bugs in production.

1. The Problem: Blocking I/O Wastes Time

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 because each call blocks

Ten requests, each taking about 250 ms on the network, run 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 say "wait, but let other work run".

Threads can do this, but they are 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, and so on. When a timer rings, the chef takes 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. Walking away from an oven to do something else is await.

Two rules you cannot break:

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 a normal 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", because the body did not run

Calling an async def function does not execute it. It returns a coroutine object: a description of work that the event loop can run. Python also prints RuntimeWarning: coroutine 'hello' was never awaited, which says the same thing: the body never ran. There are two ways to run it:

python

import asyncio

asyncio.run(hello())                 # entry point: makes a loop, runs hello() to the end

# 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 · re-runs 1 earlier block on this page first

# 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()')

# Earlier blocks on this page that this one builds on, re-run quietly first.
_pm_earlier = [
    'async def hello():\n    return "hi"\n\nprint(hello())                       # <coroutine object hello at 0x...>\n                                     # NOT "hi", because the body did not run',
]
def _pm_run_earlier(_srcs, _g):
    import sys as _s, io as _io, warnings as _w
    for _src in _srcs:
        _o, _e = _s.stdout, _s.stderr
        _s.stdout = _s.stderr = _io.StringIO()
        try:
            with _w.catch_warnings():
                _w.simplefilter('ignore')
                exec(compile(_src, '<earlier block>', 'exec'), _g)
        except (Exception, SystemExit):
            pass
        finally:
            _s.stdout, _s.stderr = _o, _e
    if 'matplotlib.pyplot' in _s.modules:
        _s.modules['matplotlib.pyplot'].close('all')   # their plots belong to their own blocks
_pm_run_earlier(_pm_earlier, globals())
del _pm_earlier, _pm_run_earlier

asyncio.run() is the standard way to start the event loop from synchronous code. Don't call it from inside another coroutine, because that would start a loop while one is already running. Use await instead.

4. The Standard Demo: Sequential vs Concurrent

python

import asyncio
import time

async def fetch(name, delay):
    print(f"  start {name}")
    await asyncio.sleep(delay)                   # gives control back to the loop while it waits
    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 wait at the same time)

asyncio.sleep is the async version of sleep. It tells the event loop "wake me in N seconds, and run something else until then." time.sleep would block the entire loop for 3 seconds and take the same total time as the sync version. Using time.sleep in async code is one of the most common async bugs.

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 finished. 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 simple batches. When one failure should cancel the rest (which is true for 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 is already scheduled on the loop. gather and TaskGroup are built on top of tasks. Useful patterns:

python

# Start many tasks, await them 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 task that does not stop on its own
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, and no task is left running on its own.

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))       # cancelled automatically when b fails
    # When the block exits, all tasks are 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 is better than plain gather:

Cancellation safety: one failure cancels the other tasks, so no forgotten tasks keep running in the background.
Scoped lifetime: the async with block guarantees every task finishes (or is cancelled) before the code after the block runs.
ExceptionGroup: if several tasks fail at the same time, you get all the errors together instead of losing some of them.

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+): a context manager that can wrap any code
try:
    async with asyncio.timeout(2.0):
        result = await fetch_slow()
        # any number of awaits can go here, and the 2s limit covers them all
except asyncio.TimeoutError:
    print("gave up after 2s")

asyncio.timeout can wrap a TaskGroup, several awaits, and async with blocks, while wait_for only wraps a single coroutine. Use timeout in new code.

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

9. `async for` and Async Generators

def + yield gives you a generator. In the same way, async def + yield gives you an async generator, which you read 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 you read paginated HTTP APIs, websockets, and server-sent events. Each yield produces one item, and at each await the coroutine can pause and let other tasks run. The ideas from generators still apply: values are produced lazily, one at a time, and the function pauses between yields. The only difference is that it can also await.

10. `async with`: Async Context Managers

This is the async version 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 will see async with on HTTP clients, database connection pools and async locks. Build your own with @contextlib.asynccontextmanager. See contextmanagers for the sync version. The async one has the same structure.

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 freezes the event loop until it ends
    return hashlib.sha256(data).hexdigest()

async def main():
    await asyncio.gather(*[hash_a_lot(blob) for blob in big_blobs])
    # NO speedup, because 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 avoids the GIL by using separate processes, and the concurrent.futures.ProcessPoolExecutor API works well 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, because that would block the loop. Send it to a thread or process pool instead:

python

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

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

# CPU-bound work: send it 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 works with the event loop, and you await it like a coroutine. The blocking call runs on a worker thread (or process), and the loop stays responsive.

The shortcut in 3.9+ is await asyncio.to_thread(blocking_fn, *args). It does the same thing with less code.

13. The Real-World Ecosystem

You will almost never write async code that talks directly to a socket. There are high-level async libraries that already handle the protocol for you.

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 has the structure of production code: connection pooling through 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. A type checker can catch this before you run the code: mypy reports it as an error (unused-coroutine), so run mypy on async code.

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

Any synchronous blocking call freezes the event loop. Every other task waits. A typical symptom: "my async server handles the same number of requests per second as the sync one." Use async replacements such as asyncio.sleep, httpx.AsyncClient and asyncpg, or run 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 inside 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 open 10,000 connections at once. The OS runs out of sockets,
    # the server's rate limits trigger, and memory can run out (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, asyncio.run cancels it when main() returns, so its work may never finish, and you get no error message about it. (If you manage the event loop yourself and close it with tasks still pending, you get a Task was destroyed but it is pending! message.) In long-running services, forgotten tasks also build up over time. 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, use 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 stop the whole 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` running at once.

    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 a random delay and an occasional 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())

This solution is a template for production code that must "start N network calls, limit how many run at once, and not let one failure stop the batch":

One shared AsyncClient: it reuses the connection pool across all tasks. Building a new client for each request is a common cause of slow async HTTP code.
async with sem: it acquires before the network call and releases afterwards, even if the call raises. You don't need to track this by hand.
Per-task try/except: it converts failures into a special value (None), so gather doesn't stop at 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), switch 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, and 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, but 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. It cancels the other tasks when one fails and raises an ExceptionGroup.
Timeouts: asyncio.wait_for (one coroutine) and asyncio.timeout() (any block). Prefer asyncio.timeout().
async for consumes async generators, and async with wraps async context managers such as connection pools, async locks and 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 can use up all sockets, hit the rate limits of remote APIs, and run your process out of memory.
The ecosystem: httpx, aiohttp, FastAPI, asyncpg, redis.asyncio. Async libraries now exist for most common tasks.

Next: Multiprocessing & Threading: what to do when one thread isn't enough or the GIL slows you down, and how to choose the right concurrency model.

1. The Problem: Blocking I/O Wastes Time

2. The Mental Model: One Thread, Many Tasks

3. async def Defines a Coroutine

4. The Standard 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`