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
- 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). - Keyword-only for options (Ch 5.1):
sorted(xs, key=..., reverse=True)cannot be called assorted(xs, f, True). Your option arguments go after*. - Accept protocols, return concrete types (Ch 18.2).
- Raise specific exceptions from one base class (Ch 7.4, 65.2), and
never return
Noneto mean "an error happened" for a function that usually returns a value. - No surprising side effects at import time (Ch 9.1): no network calls, no logging configuration, no global state changes.
- Immutability by default for values you hand out (frozen dataclasses, tuples) (Ch 2.7, 15.2).
- 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.