The code review round

Review this cache decorator.

import functools, time

def ttl_cache(seconds=300):
    def decorator(fn):
        cache = {}
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            key = str(args) + str(kwargs)
            if key in cache:
                value, ts = cache[key]
                if time.time() - ts < seconds:
                    return value
            value = fn(*args, **kwargs)
            cache[key] = (value, time.time())
            return value
        return wrapper
    return decorator

class UserService:
    @ttl_cache(seconds=60)
    def get_permissions(self, user_id):
        return self._db.fetch_permissions(user_id)

Blocking problems, in order of severity:

1. Unbounded memory. Entries are never evicted; expired entries are re-written but never removed, so the dict grows with the key space forever. In a service with millions of user ids, this is a slow OOM. Expiry is not eviction.
2. The cache retains self. The key includes str(args), and args[0] is the UserService instance — so every instance is pinned in the cache dict for the lifetime of the process. On a per-request service object, this leaks the instance and everything it references (including the DB connection).
3. str(args) is a broken key. str is not injective: ("1", 2) and (1, "2") can collide depending on repr, objects without a stable __repr__ produce a new key every call (defeating the cache entirely), and objects with a memory-address repr make the cache useless and unbounded at the same time. Use the arguments themselves in a tuple key, requiring hashability — which also gives you a clean error instead of silent wrongness.
4. No stampede protection. When a hot key expires, every concurrent caller misses and calls fn simultaneously. For fetch_permissions on a popular tenant, that is a burst of identical database queries at exactly the moment the system is busiest.
5. Not thread-safe. Check-then-act across the dict is not atomic under threads. In CPython the dict operations are individually safe, so you will not corrupt the dict, but you can compute the value several times — which is the same problem as (4).
6. Caching a permissions lookup for 60 seconds is a security decision, not a performance one. A revoked permission remains effective for up to a minute. That may be acceptable, but it must be a documented deliberate choice, and the reviewer should ask.

What is wrong with this pagination endpoint?

@app.get("/api/orders")
def list_orders(page: int = 1, per_page: int = 50, sort: str = "created_at"):
    offset = (page - 1) * per_page
    rows = db.execute(
        f"SELECT * FROM orders WHERE tenant_id = {g.tenant_id} "
        f"ORDER BY {sort} DESC LIMIT {per_page} OFFSET {offset}"
    )
    return {"orders": [dict(r) for r in rows], "total": db.execute("SELECT COUNT(*) FROM orders").scalar()}

• SQL injection via sort. tenant_id comes from the session so it is probably an integer, but sort is user-controlled string interpolation directly into the query. ?sort=created_at; DROP TABLE orders -- is the obvious version; the subtle version is ?sort=(SELECT ...) for data exfiltration through ordering. Identifiers cannot be parameterised, so the fix is an allow-list: if sort not in {"created_at", "total", "status"}: raise 400.
• No cap on per_page. ?per_page=1000000 is a free denial of service and an unbounded memory allocation.
• OFFSET deep paging. Page 10,000 makes the database scan and discard 500,000 rows. Cost grows with page depth, and rows shift under concurrent writes so users skip and duplicate records. Keyset pagination fixes both.
• SELECT * ships every column including ones you do not render, defeats covering indexes, and breaks the client silently when a column is added.
• The COUNT(*) is unfiltered — it counts every order across every tenant. That is both a wrong number and an information leak (a competitor can infer your total volume), and it is a full table scan on every page request.
• No index guarantee. ORDER BY <dynamic> DESC with WHERE tenant_id needs a composite index per sort column, or every request sorts a large set in memory.

This test passes. Why is it worthless?

def test_send_welcome_email(mocker):
    mock_client = mocker.patch("app.services.email.client")
    user = User(email="a@b.com", name="Ada")

    send_welcome_email(user)

    mock_client.send.assert_called_once()

The test asserts that a function called a method. It does not assert:

• Who the email went to. assert_called_once() passes if you send to the wrong address, which is the single most likely bug in this function and a privacy incident when it happens.
• What the email contains. Wrong template, unrendered {name} placeholder, wrong link — all pass.
• That the client is used correctly. mocker.patch replaces the object with a MagicMock that accepts any attribute and any signature. If someone renames send to send_message, the production code breaks and this test still passes, because the mock happily accepts mock.send(...). This is the deepest problem: the mock does not match the interface it is standing in for.

The improvements, in order of value:

def test_send_welcome_email(mocker):
    client = mocker.patch("app.services.email.client", autospec=True)   # signature-checked
    send_welcome_email(User(email="a@b.com", name="Ada"))

    client.send.assert_called_once_with(
        to="a@b.com",
        subject="Welcome, Ada",
        html=mocker.ANY,
    )
    body = client.send.call_args.kwargs["html"]
    assert "Ada" in body and "{" not in body          # catches unrendered templates

Better still: use a fake — an in-memory email backend that records messages — so the test exercises your real serialisation and you assert on a captured message object. Django ships exactly this (mail.outbox), and it is more robust than any mock.

Find the bug in this async batching worker.

async def process_events(queue: asyncio.Queue, batch_size: int = 100):
    batch = []
    while True:
        event = await queue.get()
        batch.append(event)
        if len(batch) >= batch_size:
            await write_to_warehouse(batch)
            batch = []

1. A partial batch is never flushed. If 99 events arrive and then traffic stops, they sit in memory indefinitely. In a low-traffic tenant, "indefinitely" means until the pod restarts — and then they are lost. You need a time-based flush as well as a size-based one: asyncio.wait_for(queue.get(), timeout=5) and flush on TimeoutError.
2. Data loss on shutdown. On SIGTERM the task is cancelled at the await, and the in-memory batch disappears. You need a try/except asyncio.CancelledError that flushes what it holds, and a shutdown sequence that stops accepting new events, drains the queue, and only then exits.
3. queue.task_done() is never called, so any await queue.join() elsewhere hangs forever.
4. One failure loses the whole batch. If write_to_warehouse raises, the exception propagates out of the loop, the task dies silently (a bare create_task with no done-callback swallows it), and the queue fills until it blocks producers or exhausts memory. You need error handling with a retry, a dead-letter path for a persistently bad batch, and — importantly — the batch must not be dropped on the floor.
5. No backpressure story. If the warehouse is slower than the producers, the queue grows without bound unless it was constructed with a maxsize. With maxsize set, producers block, which is what you want — but that needs to be a deliberate decision that the producer side is prepared for.
6. await write_to_warehouse(batch) blocks consumption for its duration; the queue backs up during every write. Whether that matters depends on rates, but it is the sort of thing worth measuring, and the fix (a second task writing while the first accumulates) has its own ordering implications.

Why does this endpoint sometimes return the wrong user's data?

class RequestContext:
    user = None                    # class attribute

@app.middleware("http")
async def add_context(request, call_next):
    RequestContext.user = await authenticate(request)
    return await call_next(request)

@app.get("/me")
async def me():
    return {"email": RequestContext.user.email}

RequestContext.user is a class attribute — one slot shared by every request in the process. With concurrent requests, request B's authenticate completes and overwrites the class attribute while request A is suspended at an await inside call_next. When A resumes and reads RequestContext.user, it gets B's user.

The nastiest property of this bug is its shape: it is invisible in development (one request at a time), invisible in most tests, and produces cross-account data disclosure under load. It is rare enough to be dismissed as a client bug and severe enough to be a reportable incident.

The fix is a genuinely request-scoped mechanism:

from contextvars import ContextVar
current_user: ContextVar[User | None] = ContextVar("current_user", default=None)

@app.middleware("http")
async def add_context(request, call_next):
    token = current_user.set(await authenticate(request))
    try:
        return await call_next(request)
    finally:
        current_user.reset(token)          # reset, or you leak context between tasks

ContextVar is copied per task, so each request — and each task it spawns — gets its own view. Better still in FastAPI: skip the ambient state entirely and pass the user as a dependency (user: Annotated[User, Depends(get_current_user)]), which makes the dependency explicit, testable and impossible to get wrong.

Two engineers disagree about this. Who is right?

# Engineer A wrote:
def get_user_orders(user_id: int) -> list[Order]:
    return Order.objects.filter(user_id=user_id).select_related("customer", "shipping_address")

# Engineer B says in review:
# "Drop select_related. We only use order.total in 90% of call sites, so you're
#  fetching two extra joins for nothing. Add it at the call site that needs it."

Both positions are defensible, which is the point of the question.

B is right in principle: a shared data-access function should not carry the union of all its callers' needs, because every caller then pays the most expensive caller's cost. Two extra joins on a list endpoint returning 200 orders is real work, and the wasted columns crowd out the buffer cache.

A is right about the failure mode: without select_related, the 10% of call sites that touch order.customer produce a silent N+1 that nobody notices until the table grows. The cost of B's approach is paid in incidents, not in code review.

The resolution is that both are treating the wrong thing as the design. The function should let the caller state its needs, and the system should make the mistake loud:

def get_user_orders(user_id: int, *, with_customer: bool = False) -> QuerySet[Order]:
    qs = Order.objects.filter(user_id=user_id)
    return qs.select_related("customer", "shipping_address") if with_customer else qs.only(
        "id", "total", "status", "created_at"
    )

…plus the part that actually prevents the bug: make lazy loading fail loudly in tests and development (raiseload("*") in SQLAlchemy, django-zen-queries or nplusone in Django), and assert query counts on hot endpoints so an accidental N+1 breaks CI rather than production.

You have inherited a 4,000-line module with no tests that everyone is afraid to touch. You need to change one behaviour in it. What do you do?

The sequence:

1. Do not rewrite it. A rewrite of code you do not understand, with no tests to define correct behaviour, replaces a known-working system with an unknown one and takes three times the estimate. This is the answer the interviewer most wants to hear you reject.
2. Find the seam. Identify the smallest boundary around the behaviour you need to change — a function, a class, an HTTP handler.
3. Write characterisation tests. Not tests of what the code should do — tests of what it currently does, including the behaviour that looks like a bug. Capture real production inputs if you can, feed them through, and snapshot the outputs. These tests are your safety net, and they are valuable even though they encode current bugs, because your job right now is to change one thing without changing anything else.
4. Make the change, keeping the diff as small as the change requires. Resist tidying adjacent code in the same commit — it makes the diff unreviewable and couples your risk to your cleanup.
5. Ship it behind a flag if the blast radius is large, and compare old and new paths in production (a shadow read: run both, log divergence, serve the old one) if it is critical.
6. Leave it better by a small, honest amount — the code you touched, not the whole file. Repeated over months this is how a module becomes maintainable; the big-bang cleanup never gets scheduled.