Frontend Labs
⋯

Part IX — Expert and professional Python

The last Part is about Python in teams and in production: designing code other people use, observing it while it runs, debugging it when it breaks, shipping it, and recognizing the patterns that cause trouble. It ends with the capstone that ties the whole series together.


67. Library and API design: making code other people can use

67.1 Principles, with the standard library as the example

  1. Make the common case short and the rare case possible. open(path) is one argument; open(path, mode, buffering, encoding, errors, newline, closefd, opener) is there when you need it. requests does the same (Ch 65.1).
  2. Keyword-only for options (Ch 5.1): sorted(xs, key=..., reverse=True) cannot be called as sorted(xs, f, True). Your option arguments go after *.
  3. Accept protocols, return concrete types (Ch 18.2).
  4. Raise specific exceptions from one base class (Ch 7.4, 65.2), and never return None to mean "an error happened" for a function that usually returns a value.
  5. No surprising side effects at import time (Ch 9.1): no network calls, no logging configuration, no global state changes.
  6. Immutability by default for values you hand out (frozen dataclasses, tuples) (Ch 2.7, 15.2).
  7. Be explicit about resources: anything holding a socket, file, or thread is a context manager (Ch 8, 16.5).

67.2 A small, well-shaped API

"""ratelimit: a token-bucket rate limiter."""
from __future__ import annotations

import threading
import time
from collections.abc import Callable
from dataclasses import dataclass, field

__all__ = ["RateLimiter", "RateLimited"]

class RateLimited(Exception):
    """Raised by RateLimiter.acquire(block=False) when no token is available."""
    def __init__(self, retry_after: float) -> None:
        super().__init__(f"rate limited; retry after {retry_after:.3f}s")
        self.retry_after = retry_after

@dataclass
class RateLimiter:
    """Allow `rate` operations per second with bursts up to `burst`. Thread-safe.

    >>> rl = RateLimiter(rate=5, burst=5)
    >>> with rl:          # blocks until a token is available
    ...     pass
    """
    rate: float
    burst: int = 1
    clock: Callable[[], float] = time.monotonic
    _tokens: float = field(init=False, repr=False)
    _last: float = field(init=False, repr=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False)

    def __post_init__(self) -> None:
        if self.rate <= 0 or self.burst < 1:
            raise ValueError("rate must be > 0 and burst >= 1")
        self._tokens, self._last = float(self.burst), self.clock()

    def acquire(self, *, block: bool = True, timeout: float | None = None) -> None:
        deadline = None if timeout is None else self.clock() + timeout
        while True:
            with self._lock:
                now = self.clock()
                self._tokens = min(self.burst, self._tokens + (now - self._last) * self.rate)
                self._last = now
                if self._tokens >= 1:
                    self._tokens -= 1
                    return
                wait = (1 - self._tokens) / self.rate
            if not block or (deadline is not None and self.clock() + wait > deadline):
                raise RateLimited(wait)
            time.sleep(wait)

    def __enter__(self) -> RateLimiter:
        self.acquire()
        return self

    def __exit__(self, *exc: object) -> None:
        return None

if __name__ == "__main__":
    import doctest
    doctest.testmod()
    rl = RateLimiter(rate=10, burst=2)
    t = time.perf_counter()
    for _ in range(12):
        rl.acquire()
    print(f"12 acquires at 10/s with burst 2: {time.perf_counter() - t:.2f}s")   # ~1.0s

Every principle above is in it: a two-name public API (__all__), a specific exception carrying the data a caller needs (retry_after), keyword-only options, validation at construction, an injectable clock for tests, thread safety documented and implemented, a context-manager form, and a doctest that keeps the docstring honest.

67.3 Evolving an API without breaking users

import warnings

def fetch(url: str, *, timeout: float = 10.0, verify: bool = True, **legacy) -> str:
    if "insecure" in legacy:                                    # an old option you are removing
        warnings.warn("fetch(insecure=...) is deprecated; use verify=False",
                      DeprecationWarning, stacklevel=2)        # stacklevel=2: point at the CALLER's line
        verify = not legacy.pop("insecure")
    if legacy:
        raise TypeError(f"unexpected arguments: {sorted(legacy)}")
    return f"GET {url} timeout={timeout} verify={verify}"

warnings.simplefilter("always")
print(fetch("https://x", insecure=False))

Follow semantic versioning: deprecate in a minor release with a warning, remove in the next major. 3.13 added @warnings.deprecated (PEP 702), which type checkers also understand. Run your test suite with -W error::DeprecationWarning to find your own use of deprecated APIs before they disappear.


Generated from real-life-python-guide.md · 8 Oct 2026 16:36