LearnAI Lab InterviewingAI Lab Coding Interview: Python Systems

⚙️HardMLOps & Deployment

AI Lab Coding Interview: Python Systems

Practice production-shaped Python coding prompts: crawlers, in-memory stores, ledgers, schedulers, parsers, rate limiters, caches, and concurrency follow-ups.

9 min read

Learning path

Step 152 of 155 in the full curriculum

Reasoning & Test-Time Compute AI Lab System Design Interview

AI Lab Coding Interview: Python Systems

Frontier AI lab coding rounds often look less like isolated puzzle drills and more like small production systems. You may get one base prompt, then a sequence of staged requirements: add TTLs, add concurrency, add cancellation, add rate limits, preserve deterministic output, or explain why your state model will not corrupt itself.

This article is the coding session for the final interview-prep section. The goal is to become fast at practical Python: clear state, small APIs, local tests, and honest concurrency invariants.

Python systems coding interview loop with clarify, implement, test, extend, add concurrency, and explain invariants — AI lab coding rounds reward staged implementation: ship a correct base version, test it, then extend the same small state model without losing invariants.

The operating model

Use this loop for every prompt:

Restate input, output, and failure behavior.
Ship version 1 with the smallest correct state model.
Add table-driven tests before adding stage 2.
Isolate shared mutable state before adding threads.
End by naming complexity, race risks, and production hardening.

Good interview code is not the most abstract code. It is code whose invariants can be defended while requirements change.

Python tools to know cold

Know these without documentation:

Need	Python building block
FIFO work queue	`collections.deque`, `queue.Queue`, `asyncio.Queue`
Counts and top errors	`collections.Counter`
LRU cache	`collections.OrderedDict`
Deadlines and TTL	`time.monotonic`, injected `now` function
Priority scheduling	`heapq`, `queue.PriorityQueue`
Thread safety	`threading.Lock`, `threading.RLock`, `threading.Condition`, `threading.Event`
Worker fanout	`concurrent.futures.ThreadPoolExecutor`, `as_completed`
Parsing	`splitlines`, `re`, explicit state machines

Do not wait for a test framework. Write a small run_tests() and use plain assert.

Prompt bank

Use these as drills. Do not memorize wording. Learn the patterns.

Prompt	Base implementation	Follow-ups
Same-host web crawler	BFS/DFS with visited set	concurrency, per-host rate limit, timeouts, cancellation
In-memory key/value DB	`set`, `get`, `delete`, `scan`	TTL, compare-and-set, transactions, snapshots
Banking ledger	accounts, deposit, withdraw, transfer	idempotency, reversals, scheduled transfers, deadlock avoidance
Task scheduler	dependency graph and ready queue	cycle detection, retries, worker pool, cancellation
Log parser	multiline event grouping	malformed lines, rolling windows, top errors
Rate limiter	fixed or sliding window	token bucket, multi-dimensional quotas, retry-after
LRU/TTL cache	capacity eviction	TTL, thread safety, metrics, stale cleanup

Drill 1: token bucket with deterministic time

Rate limiters show up because they combine state, boundary conditions, and production behavior. Retry policies often need jitter to avoid synchronized retry storms, so this prompt is really about overload control as much as counters.^[1] A strong implementation injects time so tests do not sleep.

token-bucket.py

from dataclasses import dataclass

@dataclass
class Bucket:
    capacity: float
    refill_per_second: float
    tokens: float
    updated_at: float

class TokenBucketLimiter:
    def __init__(self, capacity: int, refill_per_second: float) -> None:
        self.capacity = float(capacity)
        self.refill_per_second = float(refill_per_second)
        self._buckets: dict[str, Bucket] = {}

    def allow(self, key: str, now: float, cost: float = 1.0) -> tuple[bool, float]:
        bucket = self._buckets.get(key)
        if bucket is None:
            bucket = Bucket(self.capacity, self.refill_per_second, self.capacity, now)
            self._buckets[key] = bucket

        elapsed = max(0.0, now - bucket.updated_at)
        bucket.tokens = min(bucket.capacity, bucket.tokens + elapsed * bucket.refill_per_second)
        bucket.updated_at = now

        if bucket.tokens >= cost:
            bucket.tokens -= cost
            return True, 0.0

        missing = cost - bucket.tokens
        retry_after = missing / bucket.refill_per_second
        return False, retry_after

limiter = TokenBucketLimiter(capacity=3, refill_per_second=1.0)
print([limiter.allow("org-a", now=0.0)[0] for _ in range(4)])
print(limiter.allow("org-a", now=0.5))
print(limiter.allow("org-a", now=1.0))

Output

[True, True, True, False]
(False, 0.5)
(True, 0.0)

What to say out loud:

The key can be a user, organization, endpoint, or model.
now is injected for deterministic tests.
Cleanup for idle buckets is a production memory concern, not a correctness requirement for the base prompt.
A thread-safe version needs a lock around _buckets and bucket mutation.

Drill 2: thread-safe same-host crawler shape

A crawler prompt tests graph traversal plus concurrency. The invariant is simple: each URL is claimed once before it is fetched or enqueued.

single-process-crawler-core.py

from collections import deque
from urllib.parse import urlparse, urljoin

PAGES = {
    "https://lab.example/start": ["/a", "/b", "https://other.example/x"],
    "https://lab.example/a": ["/b", "/c"],
    "https://lab.example/b": ["/c"],
    "https://lab.example/c": [],
}

def get_urls(url: str) -> list[str]:
    return PAGES.get(url, [])

def same_host_crawl(start_url: str) -> list[str]:
    start_host = urlparse(start_url).netloc
    queue = deque([start_url])
    visited = {start_url}
    ordered: list[str] = []

    while queue:
        url = queue.popleft()
        ordered.append(url)
        for raw_link in get_urls(url):
            link = urljoin(url, raw_link)
            if urlparse(link).netloc != start_host:
                continue
            if link in visited:
                continue
            visited.add(link)
            queue.append(link)

    return ordered

print(same_host_crawl("https://lab.example/start"))

Output

['https://lab.example/start', 'https://lab.example/a', 'https://lab.example/b', 'https://lab.example/c']

Concurrency follow-up:

Protect visited with a lock.
Claim a URL while holding the lock, before scheduling a worker.
Keep output nondeterministic unless the prompt explicitly asks for deterministic order.
Add timeouts and failed-fetch handling without retry storms.

Drill 3: ledger with idempotency

Ledger prompts test whether you can keep money-like state consistent. Use append-only events when possible; if you maintain balances, update balance and event together.

ledger-idempotency.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Event:
    idempotency_key: str
    account: str
    delta: int

class Ledger:
    def __init__(self) -> None:
        self.balance: dict[str, int] = {}
        self.events: list[Event] = []
        self.seen: set[str] = set()

    def apply(self, key: str, account: str, delta: int) -> int:
        if key in self.seen:
            return self.balance.get(account, 0)
        new_balance = self.balance.get(account, 0) + delta
        if new_balance < 0:
            raise ValueError("insufficient funds")
        self.balance[account] = new_balance
        self.events.append(Event(key, account, delta))
        self.seen.add(key)
        return new_balance

ledger = Ledger()
print(ledger.apply("deposit-1", "acct", 100))
print(ledger.apply("deposit-1", "acct", 100))
print(ledger.apply("withdraw-1", "acct", -30))
print(ledger.balance["acct"], len(ledger.events))

Output

Transfer follow-up:

Use one idempotency key for the whole transfer.
Lock account IDs in sorted order to avoid deadlock.
Record both debit and credit events together.
Define whether external side effects happen before or after durable commit.

Concurrency answer template

When asked to make a solution concurrent, say this before writing code:

Shared state is X.
The lock protects X.
A work item is claimed at this point.
Worker shutdown happens through this sentinel, event, or executor lifecycle.
Failed work records an error and does not corrupt shared state.

This sounds mechanical because it should. Concurrency interview failures usually come from vague ownership.

Common pitfalls

Solving version 1, then adding TTL, transactions, or threads without restating the invariant.
Using wall-clock sleeps in tests instead of injected time.
Making crawler output order part of correctness after adding concurrent workers.
Treating idempotency as "retry the operation" instead of recording the request key and result semantics.
Adding a global lock around everything without explaining throughput, deadlock, and fairness tradeoffs.

Mastery checklist

Implement a rate limiter with deterministic time and boundary tests.
Implement a same-host crawler and explain where to place the visited lock.
Implement a ledger with idempotency and insufficient-funds behavior.
Explain how to add TTLs to a key/value store without sleeping in tests.
Build a dependency scheduler with cycle detection and a ready queue.
Parse multiline logs with malformed-line handling.
State complexity and memory growth for each solution.

Next Step

Continue to AI Lab System Design Interview

You will turn the same building blocks into end-to-end AI/backend designs with scale, overload behavior, permissions, rollout gates, and observability.

PreviousReasoning & Test-Time Compute

Share this article

X Facebook LinkedIn Bluesky Reddit Hacker News Email

References

Exponential Backoff And Jitter

Brooker, M. (AWS) · 2015

Back to Topics

LearnAI Lab InterviewingAI Lab Coding Interview: Python Systems

⚙️HardMLOps & Deployment

AI Lab Coding Interview: Python Systems

Practice production-shaped Python coding prompts: crawlers, in-memory stores, ledgers, schedulers, parsers, rate limiters, caches, and concurrency follow-ups.

9 min read

Learning path

Step 152 of 155 in the full curriculum

Reasoning & Test-Time Compute AI Lab System Design Interview

AI Lab Coding Interview: Python Systems

This article is the coding session for the final interview-prep section. The goal is to become fast at practical Python: clear state, small APIs, local tests, and honest concurrency invariants.

The operating model

Use this loop for every prompt:

Restate input, output, and failure behavior.
Ship version 1 with the smallest correct state model.
Add table-driven tests before adding stage 2.
Isolate shared mutable state before adding threads.
End by naming complexity, race risks, and production hardening.

Good interview code is not the most abstract code. It is code whose invariants can be defended while requirements change.

Python tools to know cold

Know these without documentation:

Need	Python building block
FIFO work queue	`collections.deque`, `queue.Queue`, `asyncio.Queue`
Counts and top errors	`collections.Counter`
LRU cache	`collections.OrderedDict`
Deadlines and TTL	`time.monotonic`, injected `now` function
Priority scheduling	`heapq`, `queue.PriorityQueue`
Thread safety	`threading.Lock`, `threading.RLock`, `threading.Condition`, `threading.Event`
Worker fanout	`concurrent.futures.ThreadPoolExecutor`, `as_completed`
Parsing	`splitlines`, `re`, explicit state machines

Do not wait for a test framework. Write a small run_tests() and use plain assert.

Prompt bank

Use these as drills. Do not memorize wording. Learn the patterns.

Prompt	Base implementation	Follow-ups
Same-host web crawler	BFS/DFS with visited set	concurrency, per-host rate limit, timeouts, cancellation
In-memory key/value DB	`set`, `get`, `delete`, `scan`	TTL, compare-and-set, transactions, snapshots
Banking ledger	accounts, deposit, withdraw, transfer	idempotency, reversals, scheduled transfers, deadlock avoidance
Task scheduler	dependency graph and ready queue	cycle detection, retries, worker pool, cancellation
Log parser	multiline event grouping	malformed lines, rolling windows, top errors
Rate limiter	fixed or sliding window	token bucket, multi-dimensional quotas, retry-after
LRU/TTL cache	capacity eviction	TTL, thread safety, metrics, stale cleanup

Drill 1: token bucket with deterministic time

token-bucket.py

from dataclasses import dataclass

@dataclass
class Bucket:
    capacity: float
    refill_per_second: float
    tokens: float
    updated_at: float

class TokenBucketLimiter:
    def __init__(self, capacity: int, refill_per_second: float) -> None:
        self.capacity = float(capacity)
        self.refill_per_second = float(refill_per_second)
        self._buckets: dict[str, Bucket] = {}

    def allow(self, key: str, now: float, cost: float = 1.0) -> tuple[bool, float]:
        bucket = self._buckets.get(key)
        if bucket is None:
            bucket = Bucket(self.capacity, self.refill_per_second, self.capacity, now)
            self._buckets[key] = bucket

        elapsed = max(0.0, now - bucket.updated_at)
        bucket.tokens = min(bucket.capacity, bucket.tokens + elapsed * bucket.refill_per_second)
        bucket.updated_at = now

        if bucket.tokens >= cost:
            bucket.tokens -= cost
            return True, 0.0

        missing = cost - bucket.tokens
        retry_after = missing / bucket.refill_per_second
        return False, retry_after

limiter = TokenBucketLimiter(capacity=3, refill_per_second=1.0)
print([limiter.allow("org-a", now=0.0)[0] for _ in range(4)])
print(limiter.allow("org-a", now=0.5))
print(limiter.allow("org-a", now=1.0))

Output

[True, True, True, False]
(False, 0.5)
(True, 0.0)

What to say out loud:

The key can be a user, organization, endpoint, or model.
now is injected for deterministic tests.
Cleanup for idle buckets is a production memory concern, not a correctness requirement for the base prompt.
A thread-safe version needs a lock around _buckets and bucket mutation.

Drill 2: thread-safe same-host crawler shape

A crawler prompt tests graph traversal plus concurrency. The invariant is simple: each URL is claimed once before it is fetched or enqueued.

single-process-crawler-core.py

from collections import deque
from urllib.parse import urlparse, urljoin

PAGES = {
    "https://lab.example/start": ["/a", "/b", "https://other.example/x"],
    "https://lab.example/a": ["/b", "/c"],
    "https://lab.example/b": ["/c"],
    "https://lab.example/c": [],
}

def get_urls(url: str) -> list[str]:
    return PAGES.get(url, [])

def same_host_crawl(start_url: str) -> list[str]:
    start_host = urlparse(start_url).netloc
    queue = deque([start_url])
    visited = {start_url}
    ordered: list[str] = []

    while queue:
        url = queue.popleft()
        ordered.append(url)
        for raw_link in get_urls(url):
            link = urljoin(url, raw_link)
            if urlparse(link).netloc != start_host:
                continue
            if link in visited:
                continue
            visited.add(link)
            queue.append(link)

    return ordered

print(same_host_crawl("https://lab.example/start"))

Output

['https://lab.example/start', 'https://lab.example/a', 'https://lab.example/b', 'https://lab.example/c']

Concurrency follow-up:

Protect visited with a lock.
Claim a URL while holding the lock, before scheduling a worker.
Keep output nondeterministic unless the prompt explicitly asks for deterministic order.
Add timeouts and failed-fetch handling without retry storms.

Drill 3: ledger with idempotency

Ledger prompts test whether you can keep money-like state consistent. Use append-only events when possible; if you maintain balances, update balance and event together.

ledger-idempotency.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Event:
    idempotency_key: str
    account: str
    delta: int

class Ledger:
    def __init__(self) -> None:
        self.balance: dict[str, int] = {}
        self.events: list[Event] = []
        self.seen: set[str] = set()

    def apply(self, key: str, account: str, delta: int) -> int:
        if key in self.seen:
            return self.balance.get(account, 0)
        new_balance = self.balance.get(account, 0) + delta
        if new_balance < 0:
            raise ValueError("insufficient funds")
        self.balance[account] = new_balance
        self.events.append(Event(key, account, delta))
        self.seen.add(key)
        return new_balance

ledger = Ledger()
print(ledger.apply("deposit-1", "acct", 100))
print(ledger.apply("deposit-1", "acct", 100))
print(ledger.apply("withdraw-1", "acct", -30))
print(ledger.balance["acct"], len(ledger.events))

Output

Transfer follow-up:

Use one idempotency key for the whole transfer.
Lock account IDs in sorted order to avoid deadlock.
Record both debit and credit events together.
Define whether external side effects happen before or after durable commit.

Concurrency answer template

When asked to make a solution concurrent, say this before writing code:

Shared state is X.
The lock protects X.
A work item is claimed at this point.
Worker shutdown happens through this sentinel, event, or executor lifecycle.
Failed work records an error and does not corrupt shared state.

This sounds mechanical because it should. Concurrency interview failures usually come from vague ownership.

Common pitfalls

Solving version 1, then adding TTL, transactions, or threads without restating the invariant.
Using wall-clock sleeps in tests instead of injected time.
Making crawler output order part of correctness after adding concurrent workers.
Treating idempotency as "retry the operation" instead of recording the request key and result semantics.
Adding a global lock around everything without explaining throughput, deadlock, and fairness tradeoffs.

Mastery checklist

Implement a rate limiter with deterministic time and boundary tests.
Implement a same-host crawler and explain where to place the visited lock.
Implement a ledger with idempotency and insufficient-funds behavior.
Explain how to add TTLs to a key/value store without sleeping in tests.
Build a dependency scheduler with cycle detection and a ready queue.
Parse multiline logs with malformed-line handling.
State complexity and memory growth for each solution.

Next Step

Continue to AI Lab System Design Interview

You will turn the same building blocks into end-to-end AI/backend designs with scale, overload behavior, permissions, rollout gates, and observability.

PreviousReasoning & Test-Time Compute

Share this article

X Facebook LinkedIn Bluesky Reddit Hacker News Email

References

Exponential Backoff And Jitter

Brooker, M. (AWS) · 2015

AI Lab Coding Interview: Python Systems

AI Lab Coding Interview: Python Systems

The operating model

What is the common failure mode in staged coding rounds?

Python tools to know cold

Why inject now instead of calling time.sleep() in a TTL or rate-limit interview test?

Prompt bank

Drill 1: token bucket with deterministic time

Drill 2: thread-safe same-host crawler shape

Drill 3: ledger with idempotency

What invariant should you say before adding threads to a crawler, scheduler, or cache?

Concurrency answer template

Common pitfalls

Mastery checklist

AI Lab Coding Interview: Python Systems

AI Lab Coding Interview: Python Systems

The operating model

What is the common failure mode in staged coding rounds?

Python tools to know cold

Why inject now instead of calling time.sleep() in a TTL or rate-limit interview test?

Prompt bank

Drill 1: token bucket with deterministic time

Drill 2: thread-safe same-host crawler shape

Drill 3: ledger with idempotency

What invariant should you say before adding threads to a crawler, scheduler, or cache?

Concurrency answer template

Common pitfalls

Mastery checklist