python · level 9

Decorators & Async

@decorators, async / await, and the event loop.

150 XP

Decorators & Async

The two Python features that look most magical at first sight, and stop looking magical the moment you see how they work.

Decorators

A decorator is a function that takes a function and returns a function. The @decorator syntax is sugar for name = decorator(name):

@my_decorator
def greet(): ...

# is identical to:

def greet(): ...
greet = my_decorator(greet)

The simplest useful decorator wraps a function to add behaviour around it:

import time

def timed(fn):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = fn(*args, **kwargs)
        elapsed = time.time() - start
        print(f"{fn.__name__}: {elapsed:.3f}s")
        return result
    return wrapper

@timed
def fetch_users():
    ...

@timed reassigns fetch_users to the wrapper. Calling fetch_users() now also prints timing.

Preserving metadata

The naïve wrapper above changes the wrapped function's __name__ and docstring. Use functools.wraps to forward them:

from functools import wraps

def timed(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        ...
    return wrapper

Without @wraps, fetch_users.__name__ would be "wrapper" — confusing in tracebacks and tooling.

Decorators with arguments

A decorator that takes arguments is a function that returns a decorator:

def retry(times):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(times):
                try:
                    return fn(*args, **kwargs)
                except Exception:
                    if attempt == times - 1:
                        raise
        return wrapper
    return decorator

@retry(times=3)
def call_api():
    ...

Three layers: outer takes the args, middle takes the function, inner is the actual wrapped behaviour.

Built-in decorators worth knowing

@staticmethod, @classmethod — class method variants.
@property — turn a method into a read-only attribute.
@dataclass — auto-generate __init__/__repr__/__eq__.
@functools.cache (3.9+) / @lru_cache — memoise.
@functools.singledispatch — function overloading by argument type.
@contextlib.contextmanager — turn a function into a with-able.

async / await

Python's coroutines. The mental model: an async def returns a coroutine object that runs cooperatively under an event loop. await pauses the current coroutine until another one finishes:

import asyncio

async def fetch(url):
    print(f"start {url}")
    await asyncio.sleep(1)            # like time.sleep, but yields
    print(f"done {url}")
    return f"<body of {url}>"

async def main():
    results = await asyncio.gather(
        fetch("https://a.com"),
        fetch("https://b.com"),
        fetch("https://c.com"),
    )
    return results

asyncio.run(main())

gather runs three coroutines concurrently. Total wall-time is ~1 second, not ~3 — they all sleep simultaneously.

Sync inside async (and vice versa)

You can't call async from sync without an event loop:

async def fetch(): ...

fetch()                 # returns a coroutine — does NOT run it
asyncio.run(fetch())    # runs it

You can call sync from async, but blocking calls (e.g. time.sleep, requests.get, file I/O without aiofiles) freeze the entire event loop. Use asyncio.to_thread(blocking_fn) to run blocking work in a thread:

result = await asyncio.to_thread(some_blocking_function, arg)

When to use async

Lots of I/O wait time (HTTP requests, database queries, file reads from many sources).
Long-running connections (websockets, server-sent events).
High concurrency for low-cost work (API server handling thousands of in-flight requests).

When NOT to use async:

CPU-bound work — async doesn't make CPU faster. Use multiprocessing.
Single-shot scripts that block once. Just call the sync version.
Code that has to interop heavily with sync libraries you don't control.

await blocks the current coroutine, not the whole program

A subtle point. While one coroutine awaits, the event loop runs others:

async def slow():
    await asyncio.sleep(2)
    return "slow"

async def fast():
    await asyncio.sleep(0.5)
    return "fast"

async def main():
    a = asyncio.create_task(slow())
    b = asyncio.create_task(fast())
    print(await b)            # "fast" after 0.5s
    print(await a)            # "slow" after 2s total

create_task schedules a coroutine to run NOW. await waits for a specific one. Without create_task they'd run sequentially.