12 · Coding Round

Exercises drawn from this codebase. Try each before reading the solution — the point is to be able to write these, not recognise them.

1 · Strip a code fence

Problem. Models wrap SQL inconsistently: ```sql … ```, bare ``` … ```, sometimes with prose around it, sometimes with an unbalanced fence. Extract the SQL.

def clean_sql(raw: str) -> str: ...
 
clean_sql("```sql\nSELECT 1\n```")            == "SELECT 1"
clean_sql("Here you go:\n```\nSELECT 1\n```") == "SELECT 1"
clean_sql("```sql\nSELECT 1")                 == "SELECT 1"   # unbalanced
clean_sql("SELECT 1;")                        == "SELECT 1"   # trailing ;

Solution

import re
 
_FENCE_BLOCK = re.compile(r"```(?:sql|postgresql|mysql)?\s*(.*?)\s*```",
                          re.DOTALL | re.IGNORECASE)
_LEADING_FENCE = re.compile(r"^\s*```(?:sql|postgresql|mysql)?\s*", re.IGNORECASE)
_TRAILING_FENCE = re.compile(r"\s*```\s*$")
 
 
def clean_sql(raw: str) -> str:
    text = (raw or "").strip()
    if not text:
        return ""
 
    match = _FENCE_BLOCK.search(text)          # prefer a complete block
    if match:
        text = match.group(1).strip()
    else:                                      # unbalanced: strip what's there
        text = _TRAILING_FENCE.sub("", _LEADING_FENCE.sub("", text)).strip()
 
    return text.rstrip().rstrip(";").strip()

Talk about: the naive approach (drop first and last line) mangles unfenced SQL and leaves stray backticks when the closing fence is missing. Non-greedy .*? matters when there are two blocks. The trailing semicolon must go because the connector wraps the SQL as SELECT * FROM (<sql>) _q LIMIT n.

2 · Validate read-only SQL

Problem. Given SQL, decide whether it is read-only. Reject DROP TABLE users, SELECT 1; DROP TABLE users, and WITH x AS (DELETE FROM t RETURNING *) SELECT * FROM x. Accept SELECT * FROM stores WHERE city = 'Alter Markt'.

Solution

import sqlglot
import sqlglot.expressions as exp
 
BLOCKED = (exp.Insert, exp.Update, exp.Delete, exp.Drop,
           exp.Create, exp.TruncateTable, exp.AlterTable, exp.Command)
ALLOWED_ROOTS = (exp.Select, exp.Subquery, exp.Union, exp.Except, exp.Intersect)
 
 
def is_read_only(sql: str, dialect: str = "postgres") -> bool:
    try:
        statements = [s for s in sqlglot.parse(sql, dialect=dialect) if s]
    except sqlglot.errors.ParseError:
        return False
 
    if len(statements) != 1:                       # blocks stacked statements
        return False
 
    root = statements[0]
    if not isinstance(root, ALLOWED_ROOTS):        # allow-list, not deny-list
        return False
 
    return not any(isinstance(node, BLOCKED) for node in root.walk())

Talk about: parse rather than pattern-match; allow-list the root shape; and the AST walk is what catches the data-modifying CTE, whose root node is an innocent Select. If you add a keyword regex as a backstop, strip string literals first or 'Alter Markt' fails.

3 · Retry with backoff and jitter

Problem. Retry an async operation on transient failures. Honour a Retry-After hint. Do not retry auth failures or an exhausted daily quota.

Solution

import asyncio, random, re
 
TRANSIENT = ("429", "500", "502", "503", "timeout", "connection reset")
PERMANENT = ("401", "403", "invalid api key", "permission denied")
DAILY_QUOTA = ("perdayper", "requestsperday", "insufficientquota")
RETRY_AFTER = re.compile(r"retry(?:\s|_)?(?:in|delay)['\":\s]*(\d+(?:\.\d+)?)\s*s", re.I)
MAX_DELAY = 60.0
 
 
class QuotaExhausted(RuntimeError): ...
 
 
def _is_daily_quota(e) -> bool:
    msg = str(e).lower()
    if not any(m in msg for m in ("429", "quota", "resource_exhausted")):
        return False
    squashed = re.sub(r"[^a-z]", "", msg)
    return any(m in squashed for m in DAILY_QUOTA)
 
 
async def with_retry(op, *, max_attempts=4, base_delay=2.0):
    for attempt in range(1, max_attempts + 1):
        try:
            return await op()
        except Exception as e:
            if _is_daily_quota(e):
                raise QuotaExhausted(str(e)) from e
            msg = str(e).lower()
            transient = any(m in msg for m in TRANSIENT) and not any(
                m in msg for m in PERMANENT
            )
            if attempt >= max_attempts or not transient:
                raise
            hint = RETRY_AFTER.search(str(e))
            delay = min(float(hint.group(1)), MAX_DELAY) if hint else min(
                base_delay * 2 ** (attempt - 1), MAX_DELAY
            )
            await asyncio.sleep(delay + random.uniform(0, delay * 0.25))

Talk about: three classes, not two. Jitter prevents the thundering herd. Cap the honoured delay — a provider can ask for an hour. And a daily quota looks transient but no in-process backoff clears it.

4 · Deterministic top-K with graph expansion

Problem. Given tables (each with a name, columns, and FK targets) and a question, return the most relevant table names. Follow foreign keys to include join tables. Cap the total. Output must be identical across processes.

Solution

import re
 
MAX_TABLES = 15
 
 
def _tokenize(text: str) -> set[str]:
    tokens = re.sub(r"([A-Z])", r"_\1", text).lower()
    return {t for t in re.split(r"[^a-z0-9]", tokens) if len(t) > 2}
 
 
def select_tables(question: str, tables: list[dict], top_k: int = 8) -> list[str]:
    if len(tables) <= MAX_TABLES:
        return [t["name"] for t in tables]
 
    q = _tokenize(question)
    scored = []
    for t in tables:
        score = len(_tokenize(t["name"]) & q) * 3.0
        for col in t.get("columns", []):
            score += len(_tokenize(col["name"]) & q) * 1.0
        scored.append((t["name"], score))
 
    # -score then name: ties break deterministically, not by input order
    scored.sort(key=lambda x: (-x[1], x[0]))
    selected = [name for name, score in scored[:top_k] if score > 0]
    if not selected:
        return [t["name"] for t in tables[:top_k]]
 
    seen, additions = set(selected), []
    for t in tables:
        if t["name"] not in seen:
            continue
        for col in t.get("columns", []):
            fk = col.get("foreign_table")
            if fk and fk not in seen and len(seen) + len(additions) < MAX_TABLES:
                additions.append(fk)
                seen.add(fk)
 
    return selected + sorted(additions)   # never `list(a_set)`

Talk about: return list(some_set) is the bug. Python randomises string hashing per process, so set order varies between runs — which changes the prompt and makes evals unreproducible. Sort ties by name, and sort the FK additions.

5 · Compare two result sets with tolerance

Problem. Decide whether two query results are equivalent. Row order should not matter. 10.0 and 10.000001 are equal. "10.0" from one driver equals 10.0 from another.

Solution

from collections import Counter
from decimal import Decimal
 
DEFAULT_TOLERANCE = 1e-6
 
 
def _comparable(value, tol=DEFAULT_TOLERANCE):
    if value is None:
        return None
    if isinstance(value, bool):          # before the numeric branch: bool is an int
        return value
    if isinstance(value, int | float | Decimal):
        return round(float(value) / tol) * tol
    text = str(value).strip()
    try:
        return round(float(text) / tol) * tol
    except (TypeError, ValueError):
        return text.lower()
 
 
def results_match(a: list[list], b: list[list], tol=DEFAULT_TOLERANCE) -> bool:
    if len(a) != len(b):
        return False
    norm = lambda rows: Counter(
        tuple(_comparable(v, tol) for v in row) for row in rows
    )
    return norm(a) == norm(b)

Talk about: Counter rather than set preserves duplicate rows, which matter. Quantising onto a tolerance grid makes near-equal floats hash equal — comparing with abs(x-y) < tol would not let you use a hash-based structure. And bool must be checked before int, because isinstance(True, int) is True.

6 · Run a sync function from async, and async from sync

Problem. (a) Call a blocking function from async code without stalling the loop. (b) Run a coroutine from a synchronous Celery task. (c) Explain why (b) fails inside a request handler.

Solution

# (a) blocking → async
result = await asyncio.to_thread(blocking_function, arg)
 
# (b) async → sync
def celery_task(arg):
    return asyncio.run(do_async_work(arg))
 
# (c) inside a request handler a loop is already running in this thread, so
#     asyncio.run() raises RuntimeError. Push it to a thread that has no loop:
handle = asyncio.create_task(asyncio.to_thread(celery_task, arg))
_running.add(handle)                       # strong ref: tasks are held weakly
handle.add_done_callback(_running.discard)

Talk about: asyncio.run() creates and closes a loop, so it cannot nest. The event loop holds tasks weakly — without a strong reference a fire-and-forget task can be garbage collected mid-flight. And a database engine’s pooled connections are bound to the loop that opened them, so a job on a fresh loop needs its own NullPool engine.

7 · A cache key that isolates tenants

Problem. Build a cache key for a query result. Identical questions differing only in whitespace or case should share an entry. Two tenants must never share one.

Solution

import hashlib
 
CACHE_PREFIX = "rm:query:"
 
 
def cache_key(source_id: str, question: str) -> str:
    normalized = question.strip().lower()
    digest = hashlib.sha256(f"{source_id}:{normalized}".encode()).hexdigest()[:24]
    return f"{CACHE_PREFIX}{digest}"

Talk about: source_id in the key is the security-relevant part — a cross-tenant cache hit leaks data. The prefix namespaces the keyspace so you can scan or flush by pattern. Truncating to 24 hex chars (96 bits) is fine for collision resistance here. Push back if asked to normalise more aggressively: stripping punctuation would merge “sales, by region” with “sales by region”, which is probably fine, but stemming would merge questions that differ meaningfully.

8 · SSE framing

Problem. Write an async generator producing Server-Sent Events for a streamed answer, reporting a mid-stream failure correctly.

Solution

import json
 
 
def sse(event_type: str, data: dict) -> str:
    return f"event: {event_type}\ndata: {json.dumps(data, default=str)}\n\n"
 
 
async def event_stream(question):
    try:
        yield sse("thinking", {"stage": "starting"})
        sql = await generate_sql(question)
        yield sse("sql", {"sql": sql})
        async for token in stream_answer(sql):
            yield sse("token", {"token": token})
        yield sse("complete", {"answer": "..."})
    except QuotaExhausted:
        # The status line is already sent — report in-band or not at all.
        yield sse("error", {"message": "...", "code": "provider_quota_exhausted"})
    finally:
        yield sse("done", {})

Talk about: the double newline terminates an event — a single one silently breaks the stream. default=str handles Decimal/datetime from a driver. The key insight is that once streaming begins you cannot change the HTTP status, so errors must be in-band with a machine-readable code. And done in a finally so the client is never left hanging.

What they are actually assessing

In a round like this the code is table stakes. What separates candidates is naming the failure mode before writing the fix — “sets are unordered, so this needs a sort”, “bool is a subclass of int, so check it first”, “the status line is already sent, so this has to be in-band”. Narrate that reasoning as you type.

11 · System Design Round Introduction