Practical Decorator Patterns

Registries, access control, deprecation, rate limiting and instrumentation - the decorator shapes that appear in real codebases, and how to test them.

The registry pattern

HANDLERS = {}


def handler(command):
    """Register a function under a command name."""
    def decorator(func):
        HANDLERS[command] = func
        return func              # note: the function itself is returned unchanged
    return decorator


@handler("add")
def handle_add(payload):
    return f"adding {payload}"


@handler("delete")
def handle_delete(payload):
    return f"deleting {payload}"


@handler("list")
def handle_list(payload):
    return "listing everything"


def dispatch(command, payload=None):
    func = HANDLERS.get(command)
    if func is None:
        return f"unknown command: {command}"
    return func(payload)


print(sorted(HANDLERS))
print(dispatch("add", "note-1"))
print(dispatch("archive"))

This decorator does not wrap anything. It records the function and hands it straight back. Registration decorators are extremely common: new behaviour is added by writing a function with a decorator, and nothing central needs editing.

Access control

from functools import wraps


class PermissionDenied(Exception):
    pass


CURRENT_USER = {"name": "meera", "roles": {"editor"}}


def requires(*roles):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            held = CURRENT_USER.get("roles", set())
            if not held & set(roles):
                raise PermissionDenied(
                    f"{func.__name__} needs one of {sorted(roles)}, "
                    f"you have {sorted(held)}"
                )
            return func(*args, **kwargs)
        return wrapper
    return decorator


@requires("editor", "admin")
def edit_note(note_id):
    return f"editing {note_id}"


@requires("admin")
def delete_everything():
    return "deleted"


print(edit_note("n-1"))

try:
    delete_everything()
except PermissionDenied as error:
    print(error)

Deprecation warnings

import warnings
from functools import wraps


def deprecated(reason, replacement=None):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            message = f"{func.__name__} is deprecated: {reason}"
            if replacement:
                message += f". Use {replacement} instead."
            warnings.warn(message, DeprecationWarning, stacklevel=2)
            return func(*args, **kwargs)

        wrapper.__doc__ = f"DEPRECATED. {reason}\n\n{func.__doc__ or ''}"
        return wrapper
    return decorator


@deprecated("the calculation was wrong", replacement="net_total")
def total(items):
    """Return the total."""
    return sum(items)


warnings.simplefilter("always")
print(total([1, 2, 3]))
print(total.__doc__.splitlines()[0])

stacklevel=2 makes the warning point at the caller rather than at the decorator, which is what a user needs to see.

Rate limiting

import time
from collections import deque
from functools import wraps


def rate_limit(calls, per_seconds):
    """Allow at most `calls` invocations in any `per_seconds` window."""

    def decorator(func):
        history = deque()

        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.monotonic()
            while history and now - history[0] > per_seconds:
                history.popleft()
            if len(history) >= calls:
                wait = per_seconds - (now - history[0])
                raise RuntimeError(f"rate limit reached, retry in {wait:.2f}s")
            history.append(now)
            return func(*args, **kwargs)

        wrapper.reset = history.clear
        return wrapper
    return decorator


@rate_limit(calls=3, per_seconds=1)
def send_message(text):
    return f"sent: {text}"


for i in range(5):
    try:
        print(send_message(f"message {i}"))
    except RuntimeError as error:
        print(error)

time.monotonic() is used rather than time.time() because it cannot go backwards when the system clock is adjusted.

Type checking from annotations

import inspect
from functools import wraps


def enforce_types(func):
    signature = inspect.signature(func)

    @wraps(func)
    def wrapper(*args, **kwargs):
        bound = signature.bind(*args, **kwargs)
        bound.apply_defaults()
        for name, value in bound.arguments.items():
            expected = func.__annotations__.get(name)
            if expected and not isinstance(value, expected):
                raise TypeError(
                    f"{func.__name__}: {name} must be {expected.__name__}, "
                    f"got {type(value).__name__}"
                )
        return func(*args, **kwargs)
    return wrapper


@enforce_types
def repeat(text: str, times: int) -> str:
    return text * times


print(repeat("ab", 3))

try:
    repeat("ab", "3")
except TypeError as error:
    print(error)

Python does not enforce annotations at runtime. A decorator can, when the input comes from outside and the cost is worth it. Static checking, covered in the type hints note, is usually the better answer.

Result caching with expiry

import time
from functools import wraps


def cache_for(seconds):
    def decorator(func):
        store = {}

        @wraps(func)
        def wrapper(*args):
            now = time.monotonic()
            if args in store:
                value, stored_at = store[args]
                if now - stored_at < seconds:
                    return value
            value = func(*args)
            store[args] = (value, now)
            return value

        wrapper.clear = store.clear
        wrapper.size = lambda: len(store)
        return wrapper
    return decorator


@cache_for(seconds=2)
def expensive(n):
    print(f"  computing {n}")
    return n * n


print(expensive(4))       # computes
print(expensive(4))       # cached
time.sleep(2.1)
print(expensive(4))       # expired, computes again

functools.cache never expires. When the underlying value can change, an expiring cache is the safer shape.

Debug tracing

from functools import wraps

DEPTH = 0


def traced(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        global DEPTH
        arguments = ", ".join(
            [repr(a) for a in args] + [f"{k}={v!r}" for k, v in kwargs.items()]
        )
        print("  " * DEPTH + f"-> {func.__name__}({arguments})")
        DEPTH += 1
        try:
            result = func(*args, **kwargs)
        finally:
            DEPTH -= 1
        print("  " * DEPTH + f"<- {result!r}")
        return result
    return wrapper


@traced
def fib(n):
    return n if n < 2 else fib(n - 1) + fib(n - 2)


fib(4)

Applied to a recursive function, this prints the whole call tree with indentation. It is the fastest way to understand a recursion you did not write.

Testing decorated functions

from functools import wraps


def double_result(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs) * 2
    return wrapper


@double_result
def add(a, b):
    return a + b


print(add(2, 3))                    # 10 - the decorated behaviour
print(add.__wrapped__(2, 3))        # 5  - the original, thanks to @wraps

@wraps stores the original on __wrapped__, which lets a test exercise the undecorated function directly. Without it there is no way back.

import unittest


class TestAdd(unittest.TestCase):
    def test_undecorated(self):
        self.assertEqual(add.__wrapped__(2, 3), 5)

    def test_decorated(self):
        self.assertEqual(add(2, 3), 10)

    def test_identity_preserved(self):
        self.assertEqual(add.__name__, "add")

Choosing not to use a decorator

# Hard to follow: four layers of indirection on one function
@retry(3)
@rate_limit(10, 60)
@requires("admin")
@cache_for(300)
@traced
def do_something(x):
    return x


# Clearer when the behaviour is genuinely conditional
def do_something(x, *, use_cache=True):
    if use_cache and x in CACHE:
        return CACHE[x]
    ...

Decorators are best for cross cutting concerns that are the same everywhere: logging, timing, access, caching, registration. When the behaviour varies per call, a parameter is clearer than a stack of wrappers.

Common mistakes

  • Forgetting @wraps, losing __name__ and __wrapped__.
  • Sharing mutable state between decorated functions by accident - define it inside decorator, not outside.
  • Using time.time() for rate limits instead of time.monotonic().
  • Registering with a decorator in a module that is never imported, so nothing registers.
  • Stacking so many decorators that the real behaviour is unreadable.
  • Putting slow work in the decorator body, which runs at import time.

Best practices

  • Use a registration decorator to replace long dispatch ladders.
  • Keep per function state inside the decorator scope so each function gets its own.
  • Expose controls on the wrapper: clear, reset, report, counters.
  • Use stacklevel=2 in warnings so the message points at the caller.
  • Test both the decorated and the undecorated behaviour.
  • Limit a function to two or three decorators.

Practice

  1. Build a plugin registry where each plugin registers itself with a decorator.
  2. Write a decorator that logs every exception and re-raises it.
  3. Write an expiring cache decorator and prove entries expire.
  4. Write a tracing decorator and use it on a recursive function.
  5. Test a decorated function both with and without its decorator, using __wrapped__.

Conclusion

Decorators are for concerns that repeat across many functions: registering, timing, caching, guarding and warning. Keep state inside the decorator, expose controls on the wrapper, preserve identity with @wraps, and stop stacking before the function's real behaviour disappears.

Written by Lorens Mishra

Default administrator account created by the installer.

Continue reading

All Python notes →
Python

Threading

Threads let a program wait for several slow things at once. In Python they help with input and output, and cannot speed up pure computation.

Read more

Discussion

0 comments
Sign in to join the discussion.

No comments yet. Be the first to say something.