Real-World Python Engineering

What this is (and isn't) optional, max-level

This page is maximal on purpose — it's the "learn as much as possible" material for someone who wants to build agents like a professional software engineer, not just make them work. None of it is required for the course or projects. Skim the headers; go deep on whatever matches what you're building. Everything is runnable and grounded in real agent scenarios.

1 · Memory & performance at scale expert

Here's a bug that has ambushed almost every engineer at least once. You write a little script that reads a log file, filters for errors, and works beautifully on the 5 MB sample on your laptop. You ship it. A month later it runs against the real log — 12 GB of it — and the process is killed by the OS with an out-of-memory error before it prints a single line. Nothing in your logic was wrong. The mistake was readlines(): you asked Python to pull the entire file into a list in one gulp, and a 12 GB file needs 12 GB of RAM you don't have.

The fix is a shift in how you move data through your program, and it rests on one idea worth internalizing: you rarely need all the data in memory at once — you only need one piece at a time. Think of drinking from a water fountain versus filling a swimming pool and then drinking. A generator is the fountain: it hands you the next line the instant you ask, and forgets the last one. Memory stays flat — a few kilobytes — whether the file is 1 MB or 1 TB, because you're never holding more than the current line.

This section teaches three levers, in order of how often you reach for them. Generators (functions with yield) turn "load everything, then process" into "process as it streams," and they compose — you can chain a reader into a parser into a filter and no intermediate list is ever built. __slots__ is the second lever: by default every Python object carries a private dictionary so you can bolt on arbitrary attributes, and that dictionary costs real memory; when you have millions of tiny, fixed-shape records (RAG chunks, log events), declaring __slots__ drops the dict and can shrink each object by half. The third lever is profiling — and it earns its place because of the single most common mistake in this whole area: optimizing the wrong thing. Programmers have famously bad intuition about where time goes. You'll swear a loop is the bottleneck, spend an afternoon tightening it, and discover the real cost was a network call you never looked at. So the rule is: measure first, optimize second, and — as the tip below stresses — in agent code the bottleneck is almost always API latency, not your Python.

python# ✗ loads the whole file into memory
lines = open("huge.log").readlines()
errors = [l for l in lines if "ERROR" in l]

# ✓ streams line by line — constant memory, even for a 10GB file
def error_lines(path):
    with open(path) as f:
        for line in f:              # file objects are lazy iterators
            if "ERROR" in line:
                yield line             # produce one at a time

# generator pipelines compose without intermediate lists
def parse(lines):
    for l in lines:
        yield l.split("|", 2)
first_10 = list(islice(parse(error_lines("huge.log")), 10))   # only reads what it needs

# __slots__: drop the per-instance __dict__ -> big memory + speed win for
# millions of small objects (e.g. chunk records in a RAG index)
class Chunk:
    __slots__ = ("id", "text", "vec")      # no arbitrary attributes, less RAM
    def __init__(self, id, text, vec):
        self.id, self.text, self.vec = id, text, vec

2 · Resource & connection management expert

A support bot ran fine in the demo and fell over the moment real traffic hit it. Every incoming question created a fresh Anthropic() client "to be safe," each one opening its own pool of network connections. Under a hundred concurrent users the machine ran out of file descriptors — the operating system's hard cap on open connections — and new requests started failing with cryptic Too many open files errors. The code was correct in isolation and lethal at scale. The lesson: some things aren't free to keep creating, and forgetting to release them is a slow-motion crash.

The mental model that fixes this is a resource is something you borrow, and anything you borrow you must give back. A database connection, an HTTP client's connection pool, an open file, a lock — each is a finite thing the system lends you, and each will run out if you take more than you return. Picture a library with 50 copies of a book: as long as readers return them, everyone gets served; if a few people keep borrowing and never return, the shelf empties and the next reader is stuck. Leaked connections are exactly those unreturned books, and "under load" is just "lots of readers at once" — which is why leaks hide in testing and only bite in production.

Two rules cover almost everything. Rule one: reuse, don't recreate. Expensive-to-open, cheap-to-reuse resources — API clients above all — should be created once, at module load, and shared; the SDK pools connections behind that single client so a thousand requests ride the same warm pool. The common mistake is the well-meaning "fresh client per request" you saw in the story — it feels safer but it's the leak. Rule two: guarantee cleanup, even when things explode. This is what the with statement is for. A @contextmanager built on try/finally promises that the teardown runs no matter what — normal return, early return, or an exception blowing through the middle — so a connection is never orphaned by a crash. When the number of resources is dynamic (open N files, close them all), ExitStack does the same bookkeeping for a whole list at once. The habit to build: whenever you open something that must be closed, reach for with instead of a manual .close() you have to remember to call on every exit path — because you will forget one, and it'll be the one an exception takes.

python# ✗ creating a client per request leaks connections & is slow
def bad_handler(q):
    client = Anthropic()          # new connection pool every call!
    return client.messages.create(...)

# ✓ create ONE client at module load, reuse it (the SDK pools connections)
client = Anthropic()              # module-level singleton
def good_handler(q):
    return client.messages.create(...)

# a reusable connection with guaranteed cleanup via a context manager
from contextlib import contextmanager
import sqlite3

@contextmanager
def readonly_db(path):
    con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
    try:
        yield con
    finally:
        con.close()               # ALWAYS closes, even on exception

with readonly_db("shop.db") as con:
    rows = con.execute("SELECT ...").fetchall()

# ExitStack: manage a dynamic number of resources cleanly
from contextlib import ExitStack
with ExitStack() as stack:
    files = [stack.enter_context(open(p)) for p in paths]
    ...                           # all closed when the block exits

3 · Resilience patterns — rate limiting & circuit breaker expert · production

Two 3 a.m. pages, two different lessons. The first: an agent fanned out fifty tool calls in a burst, blew straight past the API's 10-requests-per-second quota, and got every call rejected with 429 Too Many Requests — the whole job failed not because anything was broken but because it asked too fast. The second, a week later: a downstream database went down, and instead of noticing, the agent retried every call, each one hanging for a 30-second timeout, threads piling up until the service itself became unresponsive. One agent was too eager; the other was too stubborn. Production code needs a guard against each.

These are genuinely different failures, and it's worth feeling the difference because beginners conflate them. Rate limiting is about you being too fast — it's self-restraint, keeping your own request rate under a quota that's working perfectly fine. A circuit breaker is about them being down — it's giving up quickly on a dependency that has already failed, so you stop throwing good requests after bad. Retry-with-backoff from P4 handles the transient middle ground (a blip that clears in a second); these two handle the extremes.

The analogies make them stick. A rate limiter is a token bucket: imagine a bucket that drips full at a steady rate, and every call must grab a token to proceed — burst through the tokens that have accumulated, then you're forced to wait for the drip. It naturally allows a short flurry (up to the bucket size) and then smooths you down to the sustainable rate, which is exactly how real quotas behave. A circuit breaker is the electrical breaker in your house: after too many faults it trips open and cuts the circuit, and while it's open it fails instantly — the whole point is that failing in a microsecond is vastly cheaper than waiting 30 seconds for a timeout, both for your latency and for the struggling dependency you'd otherwise keep hammering. The common mistake with breakers is forgetting the recovery half: a breaker that trips and never resets is just an outage, so it needs a cooldown after which it lets a trial call through to test whether the dependency has healed. Both patterns below are short, thread-aware where it matters, and drop straight into real agent code.

ratelimit.pyimport time, threading

class RateLimiter:
    """Allow `rate` calls per second (token bucket). Thread-safe."""
    def __init__(self, rate: float, burst: int):
        self.rate, self.capacity = rate, burst
        self.tokens = burst
        self.updated = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self):
        with self.lock:
            now = time.monotonic()
            self.tokens = min(self.capacity,
                              self.tokens + (now - self.updated) * self.rate)
            self.updated = now
            if self.tokens < 1:
                sleep = (1 - self.tokens) / self.rate
                time.sleep(sleep)         # wait for a token
                self.tokens = 0
            else:
                self.tokens -= 1

limiter = RateLimiter(rate=10, burst=20)   # 10/s, bursts to 20
def call(...): limiter.acquire(); return client.messages.create(...)

breaker.pyimport time

class CircuitBreaker:
    """Trip open after N consecutive failures; recover after a cooldown."""
    def __init__(self, threshold=5, cooldown=30):
        self.threshold, self.cooldown = threshold, cooldown
        self.failures, self.opened_at = 0, None

    def call(self, fn, *a, **k):
        if self.opened_at and time.monotonic() - self.opened_at < self.cooldown:
            raise RuntimeError("circuit open — failing fast")   # don't even try
        try:
            result = fn(*a, **k)
        except Exception:
            self.failures += 1
            if self.failures >= self.threshold:
                self.opened_at = time.monotonic()       # trip open
            raise
        self.failures, self.opened_at = 0, None       # success resets
        return result

4 · Configuration & logging, done right production

The worst production incident is the one that could have been caught at startup. A team deployed on a Friday; the app booted cleanly, health checks went green, everyone went home. The first user request that actually called the model failed — the API key env var had a typo, so os.environ.get("ANTHROPIC_API_KEY") quietly returned None, and that None traveled halfway through the request before erupting as a baffling TypeError deep inside the SDK. Hours of Friday-night debugging for a missing character. Config and logging are the unglamorous plumbing that decides whether your 3 a.m. self can figure out what went wrong in two minutes or two hours.

Take configuration first. The instinct is to sprinkle os.environ.get("...") wherever you happen to need a value, and it works — right up until it doesn't. The problems are subtle: env vars are always strings, so "1024" is text until you remember to int() it (and one forgotten conversion is a bug); a missing value returns None instead of complaining, so the failure surfaces far from its cause; and nobody can look in one place to see what the app even needs to run. The fix is a single typed settings object validated once at boot. Think of it as a bouncer at the door: every required value is checked and type-converted the moment the program starts, and if anything's missing the app refuses to launch — with a clear message naming the exact field. Failing loud at startup is a feature; the alternative is failing quietly in front of a user.

Then logging, where the common mistake is reaching for print(). Prints are fine for a script you watch by eye, but useless for a system you operate: you can't set a severity level, you can't route them, and — the real killer — you can't query them. Free-form text like print(f"call for {user} took {ms}ms") reads fine to a human and is invisible to a machine; there's no reliable way to later ask "what's the average latency per model today?" across a million such lines. The professional move is structured logging: emit each event as a JSON object with named fields (a trace id, the model, token counts, a duration) so a log dashboard can filter, aggregate, and alert on them. One JSON line per request is the difference between logs you scroll past and logs you can actually investigate. The two labs below build exactly these — typed settings with pydantic, and a JSON log formatter — the same pair every production team ends up with.

config.pyfrom pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="AGENT_")
    anthropic_api_key: str                     # required — fails fast if missing
    model: str = "claude-opus-4-8"
    max_tokens: int = 1024
    confidence_threshold: float = 0.6
    rung: str = "observe"

settings = Settings()      # validated once at startup; typed everywhere after

logging_setup.pyimport logging, json, sys

class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "ts": self.formatTime(record),
            "level": record.levelname,
            "msg": record.getMessage(),
            **getattr(record, "extra_fields", {}),   # attach trace_id, tokens...
        })

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
log = logging.getLogger("agent")

# one structured line per request -> queryable in any log tool
log.info("llm_call", extra={"extra_fields":
    {"trace_id": "abc", "model": "opus", "out_tokens": 120}})

5 · Real pytest — fixtures, parametrize, async, coverage essential for shipping

There's a moment every project hits where "it works on my machine" stops being good enough. You change one line to fix a bug, and something three files away silently breaks — you don't find out until a user does. Tests are the safety net that turns "I hope I didn't break anything" into "I know I didn't, because the suite is green." The engineers who move fastest aren't the ones who skip tests; they're the ones whose tests let them change code fearlessly, because a red bar catches the mistake in seconds instead of in production.

The mistake worth naming up front is treating tests as a chore you bolt on at the end — a folder of near-duplicate functions you copy-paste and dread maintaining. Real pytest is the opposite: a small set of tools that make tests cheap to write and read. The four that carry most of the weight, in plain terms: a fixture is shared setup you write once and any test can ask for by name (here, a fake model client so tests never spend money or hit the network); parametrize runs the same test body over a table of inputs, so a golden set of examples becomes a dozen tests from one function instead of a dozen copies; pytest.raises lets you assert that bad input is correctly rejected — testing the sad path, which is where most real bugs live; and markers are labels that let you slice the suite, most importantly to separate fast, free, key-less unit tests from slow, paid, model-calling evals.

That last split is the one insight that makes testing an agent practical rather than painful, so it's worth stating plainly: most of your logic doesn't need the model at all. Retrieval ranking, SQL guards, parsing, routing — you can test all of it with a mocked client, deterministically, in milliseconds, on every single commit at zero cost. The handful of tests that genuinely need to call Claude — "is the answer actually good?" — get tagged @pytest.mark.eval and run on pull requests or nightly, when spending a few cents and a few seconds is fine. Blur that line and you get the worst of both worlds: a suite too slow and expensive to run often, so nobody runs it, so it rots. Keep it sharp — pytest -m "not eval" in CI, the eval suite on demand — and you get a fast green bar you trust and a quality gate that still catches regressions in the model's actual behaviour. The lab builds the whole toolkit on top of the "mock the model" technique from P5 §8.

conftest.py + test_agent.pyimport pytest, types
from unittest.mock import MagicMock

# a fixture: reusable setup, injected by name into any test
@pytest.fixture
def mock_client():
    c = MagicMock()
    c.messages.create.return_value = types.SimpleNamespace(
        content=[types.SimpleNamespace(type="text", text="ok")],
        stop_reason="end_turn")
    return c

# parametrize: run the SAME test over many inputs (a mini eval-in-tests)
@pytest.mark.parametrize("query,expected_kb", [
    ("reset password", "kb-1"),
    ("refund window", "kb-2"),
    ("enable 2fa", "kb-3"),
])
def test_retrieval(query, expected_kb):
    from agent import kb
    _, ids = kb.context_for(query)
    assert expected_kb in ids     # 3 tests from one function

# testing exceptions
def test_bad_sql_is_blocked():
    from agent import db
    with pytest.raises(ValueError, match="blocked"):
        db.run_query("DROP TABLE orders")

# async tests (needs pytest-asyncio)
@pytest.mark.asyncio
async def test_async_path():
    result = await some_async_agent("q")
    assert result

# markers to separate fast unit tests from slow eval tests
@pytest.mark.eval          # run with: pytest -m eval   (needs an API key)
def test_real_quality(): ...

terminalpytest -v                      # run all
pytest -m "not eval"            # fast, key-free tests only (CI on every commit)
pytest --cov=agent             # coverage report (needs pytest-cov)
pytest -x -q                   # stop at first failure, quiet

6 · Packaging your agent as an installable CLI production

You built something genuinely useful — and then watched a teammate struggle for twenty minutes just to run it. "Which folder do I cd into? Do I run python main.py or python -m agent? Why does it only work from the repo root? What are all these ModuleNotFoundErrors?" The tool is fine; the packaging is the barrier. A folder of scripts is something only its author can run confidently. A package is something anyone can install with one command and invoke by name from anywhere — and that gap is what separates a personal hack from a tool your team adopts.

The idea to hold onto is that installing turns your code from a location into a name. Right now your agent lives at a path — it only runs if you're standing in the right directory with the right files around you. Packaging registers it with Python itself, so pip install copies (or links) it into the environment and, crucially, creates a command: after installing, typing devops-agent "..." in any directory just works, the same way git or ruff do. The single file that makes this happen is pyproject.toml — the modern, standardized manifest that declares three things: what your project is (name, version, the Python it needs), what it depends on (the libraries pip should pull in), and what command it exposes (the [project.scripts] entry point that maps a command name to a function to call).

Two details are worth flagging because they trip up newcomers. The first is the entry point: a line like devops-agent = "devops_agent.cli:main" reads as "the command devops-agent should run the main function inside devops_agent/cli.py" — that colon-separated module:function form is the whole mechanism behind every CLI tool you've ever installed. The second is the -e (editable) flag, and skipping it is the classic mistake: a plain pip install . copies a snapshot of your code, so every edit means reinstalling before you can test it — maddening during development. pip install -e . instead points the install at your source folder, so you edit a file and re-run the command instantly, no reinstall. That editable install is simply how you develop a package day to day; the non-editable install is for when you actually ship it. The lab shows the full pyproject.toml plus the two-line install-and-run, including uv, the fast modern alternative to pip/venv.

pyproject.toml[project]
name = "devops-agent"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = ["anthropic>=0.40", "pydantic>=2", "python-dotenv"]

[project.scripts]
devops-agent = "devops_agent.cli:main"    # `devops-agent` on the command line

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

terminalpip install -e .          # editable install
devops-agent "checkout-api is crashing"   # now a real command

# modern, fast alternative to pip/venv:
uv venv && uv pip install -e .            # uv — much faster

7 · The code-quality toolchain production

The most exhausting code review is the one that argues about spaces. A pull request comes in; three comments are about a real bug and fifteen are about indentation, import order, an unused variable, and whether there should be a blank line here. Everyone's time drains into bikeshedding, and the actual logic gets a fraction of the attention it deserves. The fix isn't discipline — it's automation. A quality toolchain is a set of small, fast programs that enforce the boring stuff mechanically, so humans get to spend their judgment on the things that actually need judgment.

The mental frame is that these tools form a gate every change passes through before it can merge, and each guards a different kind of mistake — they don't overlap, they layer. A linter/formatter (ruff, now the fast standard that replaces black+flake8+isort in one tool) settles style and catches obvious slips like unused imports — end of debate, the formatter decides. A type checker (mypy or pyright) reads the type hints you wrote in P5 and proves you're not passing a str where an int is expected — a whole class of bug caught without running the code. Tests plus coverage (pytest --cov) prove the behaviour is correct and show you which lines nobody exercised. And pre-commit is the piece that ties it together: it runs all of the above automatically before each git commit, so bad code can't even get committed, let alone merged.

The common mistake here is treating this as all-or-nothing and therefore doing none of it — "we'll add quality tooling when we have time," which never comes. So adopt a minimum viable gate and grow it: ruff for format and lint, mypy for types, and pytest -m "not eval" for the fast tests, all wired into pre-commit, with the paid eval suite running on pull requests. That's a couple of hours of setup that pays for itself the first week, because every one of those checks is a bug or an argument you'll never have to have manually again. It's the same CI discipline the course's Ch 5/6 gate sections describe, made concrete with the specific tools professional teams actually run — all fast enough to run on every save.

Capstone mini-project · a production-shaped agent runner putting it ALL together

Here's the difference between a tutorial and a system, in one screen. Every pattern on this page has, so far, stood alone — a rate limiter here, a breaker there, logging in its own box. That's how you learn them, but it's not how they live. In real code they all wrap the same call to the model, interleaved, and the interesting part is how they fit together without stepping on each other. This capstone is a single file that does exactly that: the skeleton a production agent's core actually has, with the demo agent logic stripped out so you can see the plumbing plainly.

Don't try to memorize it — read it as a checklist. The value isn't in reproducing the code from scratch; it's in being able to point at each part and name the pattern it represents, because that's the skill that transfers to whatever you build next. Watch how one ask() call threads through the whole stack: the breaker checks whether the dependency is even worth calling, the rate limiter throttles so you stay under quota, the retry loop absorbs transient blips with jittered backoff, a success resets the breaker and emits a structured log line with a trace id and timing, and a total failure raises a clean custom exception. Seven patterns you met separately, now cooperating on every single request.

The mental model to carry away is that this is a shape, not a specific program. The ask() method here just calls the model once; swap its body for a real agent loop — tool dispatch, RAG retrieval, a multi-step plan — and the surrounding resilience, throttling, config, and observability stay exactly the same. That's the whole point of learning these as reusable patterns rather than one-off tricks: the production skeleton is identical behind all six of the course's projects, and only the middle changes. If you can go through the code below and put a name beside every piece — reused client, token bucket, circuit breaker, retry with backoff, custom exception, structured logging, request-scoped context — then you own the production skeleton, and the rest is filling in the agent.

runner.pyimport time, random, uuid, logging, contextvars
from anthropic import Anthropic, RateLimitError, APIStatusError

log = logging.getLogger("agent")
trace_id = contextvars.ContextVar("trace_id", default="-")
client = Anthropic()                       # reused (§2)

class AgentError(Exception): ...           # custom exception (P4)

class Runner:
    def __init__(self, rate=8, breaker_threshold=5):
        self._tokens, self._rate, self._updated = rate, rate, time.monotonic()
        self._fails, self._opened = 0, None
        self._threshold = breaker_threshold

    def _rate_limit(self):                  # token bucket (§3)
        now = time.monotonic()
        self._tokens = min(self._rate, self._tokens + (now-self._updated)*self._rate)
        self._updated = now
        if self._tokens < 1:
            time.sleep((1-self._tokens)/self._rate); self._tokens = 0
        else: self._tokens -= 1

    def ask(self, prompt, *, max_retries=4):
        if self._opened and time.monotonic()-self._opened < 30:
            raise AgentError("circuit open")   # fail fast (§3)
        for attempt in range(max_retries):
            self._rate_limit()
            try:
                t0 = time.monotonic()
                r = client.messages.create(model="claude-opus-4-8",
                        max_tokens=1024, messages=[{"role":"user","content":prompt}])
                self._fails, self._opened = 0, None              # reset breaker
                log.info("ok", extra={"extra_fields":{"trace_id":trace_id.get(),
                    "ms":round((time.monotonic()-t0)*1000),
                    "out":r.usage.output_tokens}})
                return next((b.text for b in r.content if b.type=="text"), "")
            except (RateLimitError, APIStatusError) as e:
                self._fails += 1
                if self._fails >= self._threshold: self._opened = time.monotonic()
                delay = min(30, 2**attempt + random.uniform(0,1))   # backoff (P4)
                log.warning("retry", extra={"extra_fields":{"trace_id":trace_id.get(),
                    "attempt":attempt, "delay":round(delay,1)}})
                time.sleep(delay)
        raise AgentError("exhausted retries")

def handle(prompt, tenant):
    trace_id.set(str(uuid.uuid4()))        # request-scoped (§P5 contextvars)
    return Runner().ask(prompt)

The complete Python journey expert

🎯 Interview practice interview

The interview questions this topic gets asked — worked, with code. For the full pattern catalog see A9 · Big Tech AI-engineering patterns.

pythonimport time
class RateLimiter:
    def __init__(self, rate, per=1.0):
        self.rate, self.per = rate, per
        self.tokens = rate
        self.updated = time.monotonic()
    def allow(self):
        now = time.monotonic()
        self.tokens = min(self.rate,
            self.tokens + (now - self.updated) * self.rate / self.per)
        self.updated = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

pythonclass CircuitBreaker:
    def __init__(self, threshold=5):
        self.fails = 0; self.threshold = threshold; self.open = False
    def call(self, fn):
        if self.open:
            raise RuntimeError("circuit open")
        try:
            r = fn()
        except Exception:
            self.fails += 1
            if self.fails >= self.threshold: self.open = True
            raise
        self.fails = 0
        return r

def error_lines(lines):
    for line in lines:          # file objects iterate the same way
        if "ERROR" in line:
            yield line

sample = ["INFO ok\n", "ERROR auth failed\n", "ERROR 503\n", "INFO served\n"]
errs = list(error_lines(sample))
print(len(errs), "error lines")   # 2 error lines

class LogRecord:
    __slots__ = ("svc", "level", "msg")
    def __init__(self, svc, level, msg):
        self.svc, self.level, self.msg = svc, level, msg

r = LogRecord("checkout-api", "ERROR", "auth failed")
print(r.svc, r.level)
try:
    r.extra = "nope"           # not in __slots__
except AttributeError as e:
    print("blocked:", e)

class Connection:
    def __init__(self, name): self.name = name; self.open = False
    def __enter__(self):
        self.open = True
        print(f"opened {self.name}")
        return self
    def __exit__(self, exc_type, exc, tb):
        self.open = False
        print(f"closed {self.name}")
        return False              # don't swallow exceptions

try:
    with Connection("db") as c:
        print("using", c.name)
        raise RuntimeError("boom")
except RuntimeError:
    print("error handled; connection still closed")

import time, threading

class RateLimiter:
    def __init__(self, rate, burst):
        self.rate, self.capacity = rate, burst
        self.tokens = burst
        self.updated = time.monotonic()
        self.lock = threading.Lock()
    def try_acquire(self):
        with self.lock:
            now = time.monotonic()
            self.tokens = min(self.capacity,
                              self.tokens + (now - self.updated) * self.rate)
            self.updated = now
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            return False

lim = RateLimiter(rate=10, burst=3)
granted = sum(lim.try_acquire() for _ in range(10))
print("immediately granted:", granted)   # ~3 (the burst)

class CircuitOpen(Exception): pass

class CircuitBreaker:
    def __init__(self, threshold=3):
        self.threshold = threshold
        self.failures = 0
        self.open = False
    def call(self, fn, *args, **kwargs):
        if self.open:
            raise CircuitOpen("circuit is open; refusing call")
        try:
            result = fn(*args, **kwargs)
        except Exception:
            self.failures += 1
            if self.failures >= self.threshold:
                self.open = True
            raise
        self.failures = 0
        return result

def always_fails(): raise ValueError("down")

cb = CircuitBreaker(threshold=3)
for i in range(5):
    try:
        cb.call(always_fails)
    except CircuitOpen as e:
        print(i, "->", e); break
    except ValueError:
        print(i, "-> failed")

import logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("runner")

class CircuitOpen(Exception): pass

class Breaker:
    def __init__(self, threshold=3): self.t=threshold; self.f=0; self.open=False
    def call(self, fn, x):
        if self.open: raise CircuitOpen()
        try:
            r = fn(x); self.f = 0; return r
        except Exception:
            self.f += 1
            if self.f >= self.t: self.open = True
            raise

def process(x):
    if x < 0: raise ValueError("bad task")
    return x * 2

def run(tasks):
    br = Breaker(threshold=2)
    done, failed = [], 0
    for t in tasks:
        try:
            done.append(br.call(process, t))
        except CircuitOpen:
            log.warning("circuit open, stopping"); break
        except ValueError:
            failed += 1; log.error("task %s failed", t)
    return {"done": done, "failed": failed}

summary = run([1, 2, 3])
assert summary == {"done": [2, 4, 6], "failed": 0}
print(summary)

Knowledge check check yourself