Senior Coding Interview

Senior Coding Interview

Execute L6+ real-world coding interviews where the problem is building a small system, not solving an algorithm puzzle. The core differentiator at Staff+ level is not whether you can solve it, but how you solve it: clean abstractions, narrated reasoning, graceful iteration, and production sensibility.

When to Use

• Practicing real-world coding problems (in-memory stores, rate limiters, task schedulers)
• Reviewing interview code for senior-level signals
• Preparing communication strategy for live coding sessions
• Working through CodeSignal incremental-style problems
• Mock interview practice with follow-up extensions

NOT for:

• LeetCode/competitive programming (segment trees, suffix arrays, contest optimization)
• Behavioral interviews (use interview-loop-strategist)
• System design whiteboard with no code (use ml-system-design-interview)
• Resume or career strategy

---

The 4-Stage Approach

flowchart LR
    C[1. CLARIFY\n5 min] --> S[2. SKELETON\n20 min]
    S --> I[3. ITERATE\n10 min]
    I --> O[4. OPTIMIZE\n5 min]

    C -.- C1["Ask 3-5 questions\nRestate problem\nConfirm API contract\nIdentify edge cases"]
    S -.- S1["Data structures first\nPublic API methods\nCore logic\nManual test 1 case"]
    I -.- I1["Edge cases\nError handling\nFollow-up extensions\nRefactor if needed"]
    O -.- O1["Complexity analysis\nTrade-off discussion\nConcurrency mention\nScaling path"]

Stage 1: Clarify (5 minutes)

Goal: Demonstrate you think before coding. Ask questions that reveal ambiguity the interviewer planted intentionally.

Mandatory questions for every problem: 1. Scale: "How many items/requests are we expecting?" (determines data structure choice) 2. API surface: "Should this be a class with methods, or standalone functions?" 3. Constraints: "Are keys always strings? Can values be None/null?" 4. Concurrency: "Single-threaded for now, or should I consider thread safety?" 5. Error handling: "Should invalid input raise exceptions or return error values?"

Restate the problem in your own words before writing any code. This catches misunderstandings early and signals comprehension.

Stage 2: Skeleton (20 minutes)

Goal: Get a working solution for the core case. Not perfect, not optimized -- working.

Order of implementation: 1. Define the data model (dataclass or NamedTuple for structured data) 2. Write the class/function signatures with type hints 3. Implement the happy path 4. Manually trace through one example out loud

Senior signal: Start with the public API, not the internal helpers. Show top-down thinking.

Stage 3: Iterate (10 minutes)

Goal: Handle follow-ups. This is where Staff+ candidates differentiate -- each extension should feel like a natural evolution, not a rewrite.

The follow-up ladder (interviewers typically go 2-3 levels deep): 1. Working -- Base problem solved 2. Edge Cases -- Empty inputs, duplicates, overflow, None values 3. Concurrent -- Thread safety, locks, atomic operations 4. Distributed -- Multiple nodes, consistency, partitioning 5. Fault-tolerant -- Crash recovery, persistence, graceful degradation

Senior signal: When asked "how would you make this distributed?", discuss the trade-offs before changing code. Name specific patterns (consistent hashing, write-ahead logs). You don't need to implement distributed systems in 40 minutes -- you need to show you know the path.

Stage 4: Optimize & Discuss (5 minutes)

Goal: Show you understand what you built and where it breaks.

Cover:

• Time and space complexity for each operation
• What would break at 10x scale
• What you would change given more time
• Testing strategy (what tests would you write first?)

---

Problem Archetypes

---

Communication Protocol

Senior interviews are 50% code and 50% communication. The interviewer is evaluating whether they want to work with you, not just whether you can solve the problem.

What to Narrate

• Before writing: "I'm going to use a dict with timestamps as values because we need O(1) lookup and the TTL check can be lazy."
• At decision points: "I could use a heap here for O(log n) insert, but since we're told the number of items is small, a sorted list with bisect is simpler and good enough."
• When stuck: "I'm not sure about the best way to handle concurrent access here. Let me get the single-threaded version working first, then we can discuss locks."
• After completing: "The core operations are O(1) for get/set. The cleanup sweep is O(n) but only runs periodically."

What NOT to Do

• Don't narrate syntax: "Now I'm writing a for loop..." -- the interviewer can see that.
• Don't go silent for more than 60 seconds. If you're thinking, say so.
• Don't ask "Is this right?" -- instead say "Let me trace through an example to verify."

---

Senior Signals Checklist

These are the things that make an interviewer write "strong hire" for L6+:

---

Python Patterns for Senior Interviews

Senior candidates use Python idioms that signal deep experience. See references/python-patterns-senior.md for the complete catalog with examples.

Key patterns to internalize:

• @dataclass for any structured data (not raw dicts)
• Context managers for resource cleanup
• Generators for streaming/lazy evaluation
• collections.defaultdict, Counter, deque -- know the stdlib
• Type hints on public methods (skip on internal helpers in time-pressured interviews)
• Exception hierarchies for domain errors

---

Anti-Patterns

Anti-Pattern: LeetCode Brain

Novice: Reaches for algorithmically elegant solutions (segment trees, suffix arrays, Fenwick trees) when a hash map or sorted list suffices. Spends 15 minutes on optimal time complexity for a problem where n < 1000.

Expert: Chooses the simplest correct solution first. Uses built-in data structures (dict, list, deque, heapq) unless the problem explicitly demands otherwise. Discusses when algorithmic sophistication matters only if asked about scale. The goal is working, readable, maintainable code -- not a competitive programming submission.

Detection: Solution is asymptotically optimal but unmaintainable. Candidate cannot explain trade-offs between their approach and a simpler one. No working solution exists at the 25-minute mark because they're still optimizing.

Anti-Pattern: Silent Coder

Novice: Writes code for 15+ minutes without speaking. Treats the interview like a solo coding session. When they do speak, they narrate syntax ("Now I'm writing a for loop") rather than intent.

Expert: Narrates intent before writing code ("I'm going to use a dict here because we need O(1) lookup by key"). Asks clarifying questions when ambiguity appears. Signals uncertainty honestly ("I'm not sure if Python's heapq supports decrease-key -- let me use a different approach that I'm confident in"). Treats the interviewer as a collaborator, not an examiner.

Detection: Interviewer has to prompt "what are you thinking?" more than twice. Long silences followed by large code blocks. No questions asked during the clarify phase.

Anti-Pattern: Premature Optimization

Novice: Starts with the distributed/concurrent/fault-tolerant version before solving the single-machine case. Adds caching, sharding, or thread pools before there's a working solution to optimize. Designs for 10 million users when the problem says "a few thousand."

Expert: Gets a working solution first, then optimizes when asked. Separates "what I'd do in production" from "what I'm implementing in this 40-minute interview." When the interviewer asks about scale, discusses the optimization path verbally: "I'd add a write-ahead log for durability, then shard by key hash for horizontal scaling."

Detection: No working solution at the 25-minute mark. Code has Lock, ThreadPoolExecutor, or asyncio imports but no passing test case. Architecture diagram exists but core logic doesn't.

---

CodeSignal Incremental Format

CodeSignal's pre-recorded incremental format (used by Anthropic and others) differs from live interviews. See references/codesignal-incremental.md for detailed strategy.

Key differences:

• No interviewer to ask questions -- you must self-clarify from the problem statement
• Incremental stages build on your previous code -- design for extension from the start
• Time pressure is real but self-managed -- no one tells you to move on
• You can re-read the problem statement -- do it before each stage

---

Time Budget Decision Tree

flowchart TD
    START[Problem received] --> READ["Read ENTIRE problem\n(2 min)"]
    READ --> KNOWN{Recognize\nthe archetype?}
    KNOWN -->|Yes| FAST["Fast-track clarify\n(2 min)"]
    KNOWN -->|No| DEEP["Deep clarify\n(5 min)"]
    FAST --> CODE["Code skeleton\n(18 min)"]
    DEEP --> CODE
    CODE --> CHECK{Working\nsolution?}
    CHECK -->|No, 25 min mark| TRIAGE["Simplify approach\nGet SOMETHING working\n(5 min)"]
    CHECK -->|Yes| EXTEND["Handle follow-ups\n(10 min)"]
    TRIAGE --> EXTEND
    EXTEND --> WRAP["Complexity + trade-offs\n(5 min)"]

---

References

• references/problem-archetypes.md -- Consult for worked examples of 8 problem archetypes with skeletons, clarifying questions, and follow-up extensions
• references/python-patterns-senior.md -- Consult for senior Python idioms that signal experience: dataclasses, context managers, generators, stdlib mastery, testing hooks
• references/codesignal-incremental.md -- Consult when preparing for CodeSignal's pre-recorded incremental format: time management, extension strategies, self-testing without an interviewer

CodeSignal Incremental Format Strategy

Consult this file when preparing for CodeSignal's pre-recorded, IDE-based incremental coding assessment. This format is used by Anthropic, Databricks, Roblox, and others for engineering roles at L5+.

---

How It Differs from Standard Interviews

Key Implications

1. No one to ask: You must extract all constraints from the problem statement. Read it three times before coding. 2. Incremental stages: Each stage extends your previous code. If your stage 1 design is rigid, stages 2-4 become painful. 3. You can run code: Use this. Write a quick test after each stage. Don't just submit and hope. 4. Time is yours to manage: No one will say "let's move on." You must discipline yourself.

---

The 4-Stage Structure

Typical CodeSignal incremental problems follow this pattern:

Stage 1: Basic CRUD operations on a simple data structure
Stage 2: Add a query/search capability or constraint
Stage 3: Add time-based behavior, complex queries, or state management
Stage 4: Add concurrency, optimization, or a fundamentally harder extension

Example Progression (In-Memory Database)

• Stage 1: Implement set(key, field, value), get(key, field), delete(key)
• Stage 2: Add scan(key) to return all fields, scan_prefix(prefix) to find matching keys
• Stage 3: Add TTL support -- entries expire after a configurable time
• Stage 4: Add transaction support -- begin(), commit(), rollback() with isolation

Example Progression (Task Manager)

• Stage 1: Add task, complete task, get task by ID
• Stage 2: Add priorities, return tasks in priority order
• Stage 3: Add task dependencies -- task B can't start until task A completes
• Stage 4: Add recurring tasks and time-based scheduling

---

Time Management Per Stage

Total time: ~70 minutes for 4 stages.

Time Allocation Within Each Stage

1. Read the problem statement (2-3 min) -- Read it completely. Highlight constraints. Identify what changed from the previous stage. 2. Plan the extension (2-3 min) -- How does this change your existing data model? What methods need to change? What new methods are needed? 3. Implement (8-10 min) -- Write the code. Start with the data model changes, then update existing methods, then add new ones. 4. Test (2-3 min) -- Run the provided test cases. Add one edge case test of your own.

The 5-Minute Rule

If you've been stuck on a single stage for more than 5 minutes without making progress:

1. Step back: Re-read the problem statement. You probably missed a constraint. 2. Simplify: Can you solve a simpler version of this stage? 3. Skip and return: If stage 3 is blocking you, read stage 4. Sometimes stage 4 gives you insight into what stage 3 expects. 4. Partial solution: Submit what you have. Partial credit exists.

---

Design for Extension from Stage 1

The most common trap is writing stage 1 code that's hard to extend. Here's how to avoid it.

Bad: Rigid Stage 1

# This will be painful to extend with TTL, transactions, etc.
class Database:
    def __init__(self):
        self.data = {}  # key -> value (flat)

    def set(self, key, value):
        self.data[key] = value

    def get(self, key):
        return self.data.get(key)

Good: Extensible Stage 1

from dataclasses import dataclass, field
from typing import Any


@dataclass
class Record:
    """Wrap values in a record -- easy to add metadata later (TTL, version, etc.)."""
    fields: dict[str, Any] = field(default_factory=dict)


class Database:
    def __init__(self) -> None:
        self._records: dict[str, Record] = {}

    def set(self, key: str, field_name: str, value: Any) -> None:
        if key not in self._records:
            self._records[key] = Record()
        self._records[key].fields[field_name] = value

    def get(self, key: str, field_name: str) -> Any | None:
        record = self._records.get(key)
        if record is None:
            return None
        return record.fields.get(field_name)

    def delete(self, key: str) -> bool:
        if key in self._records:
            del self._records[key]
            return True
        return False

Why this is better: When stage 3 asks for TTL, you just add expires_at: float | None = None to the Record dataclass. When stage 4 asks for transactions, you can snapshot _records or layer a transaction dict on top. The Record wrapper is the key -- it's where metadata lives.

Extension Pattern: Layer, Don't Rewrite

When a new stage arrives:

1. Add fields to your dataclass (don't change the shape of _records) 2. Add a check method (e.g., _is_valid(record) that checks TTL) 3. Modify existing methods to call the check (e.g., get calls _is_valid before returning) 4. Add new methods for new operations

# Stage 3: Adding TTL -- extend Record, don't restructure
@dataclass
class Record:
    fields: dict[str, Any] = field(default_factory=dict)
    expires_at: float | None = None  # NEW

    @property
    def is_expired(self) -> bool:  # NEW
        if self.expires_at is None:
            return False
        return time.monotonic() > self.expires_at


class Database:
    # ... existing methods ...

    def set_with_ttl(self, key: str, field_name: str, value: Any, ttl: float) -> None:
        if key not in self._records:
            self._records[key] = Record(expires_at=time.monotonic() + ttl)
        record = self._records[key]
        record.fields[field_name] = value
        record.expires_at = time.monotonic() + ttl  # Reset TTL on write

    def get(self, key: str, field_name: str) -> Any | None:
        record = self._records.get(key)
        if record is None:
            return None
        if record.is_expired:  # NEW CHECK
            del self._records[key]
            return None
        return record.fields.get(field_name)

---

Common Traps

Trap 1: Breaking Earlier Stages

Problem: Your stage 3 changes cause stage 1 and 2 test cases to fail.

Prevention:

• Run ALL test cases after each stage, not just the new ones.
• Make new features opt-in: set_with_ttl is a new method, set still works without TTL.
• Use default values: expires_at: float | None = None means existing records without TTL still work.

Trap 2: Over-Engineering for Later Stages

Problem: You read all 4 stages at once, then try to build the stage 4 architecture from stage 1.

Prevention:

• Read only the current stage's problem statement.
• Build the simplest correct solution for the current stage.
• Trust that a clean, well-structured stage 1 solution will be extensible.
• If you must peek ahead, only look at stage 2 to avoid obvious design dead ends.

Trap 3: Not Running Tests

Problem: You submit code that has syntax errors or fails basic test cases because you never ran it.

Prevention:

• CodeSignal gives you a Run button. Use it after every stage.
• Write a minimal test before the provided ones: db = Database(); db.set("k", "f", "v"); assert db.get("k", "f") == "v"
• Check return types: does the problem say return None or raise an exception for missing keys?

Trap 4: Spending Too Long on Stage 1

Problem: You spend 25 minutes perfecting stage 1 and have 45 minutes for 3 harder stages.

Prevention:

• Stage 1 should take 12-15 minutes max. It's the warm-up.
• Don't add features that aren't asked for in stage 1. No TTL, no transactions, no thread safety.
• Write the dataclass + 3-4 methods + run tests. Done.

Trap 5: Copy-Paste Errors

Problem: You duplicate code between stages and introduce subtle bugs when modifying one copy but not the other.

Prevention:

• Extract shared logic into methods: _get_record(key) that handles existence check + expiry check.
• When you find yourself copying code, make a helper.

---

Self-Testing Without an Interviewer

In a live interview, the interviewer validates your understanding. In CodeSignal, you must validate yourself.

Before Writing Code

Ask yourself these questions (write answers as comments if it helps):

# Q: What are the inputs and their types?
# A: key is str, field is str, value is Any

# Q: What should happen for invalid input?
# A: Problem says "assume valid input" -- no need for validation

# Q: What should get() return for a missing key?
# A: Problem says "return empty string" -- NOT None, NOT raise

# Q: Are operations case-sensitive?
# A: Problem doesn't specify -- assume yes (case-sensitive)

After Writing Code

Run through this mental checklist:

1. Empty state: Does your code work on a fresh instance with no data? 2. Single item: Does set + get work for one item? 3. Overwrite: Does setting the same key twice keep the latest value? 4. Delete nonexistent: Does deleting a key that doesn't exist crash or return gracefully? 5. Order: If the problem mentions ordering, is your output sorted correctly?

Quick Smoke Test Template

def smoke_test():
    db = Database()

    # Basic CRUD
    db.set("user:1", "name", "Alice")
    assert db.get("user:1", "name") == "Alice"

    # Overwrite
    db.set("user:1", "name", "Bob")
    assert db.get("user:1", "name") == "Bob"

    # Missing key
    assert db.get("user:999", "name") is None  # or whatever the spec says

    # Delete
    assert db.delete("user:1") is True
    assert db.delete("user:1") is False  # Already deleted
    assert db.get("user:1", "name") is None

    print("All smoke tests passed")

smoke_test()

---

Stage-by-Stage Mindset

Stage 4 Survival Strategy

Stage 4 is intentionally hard. Many candidates don't finish it. Your goal is to demonstrate understanding, not necessarily pass all test cases.

1. Read the full problem statement -- understand what's being asked 2. Write the data model changes -- show you know what the solution looks like 3. Implement the core path -- handle the happy case 4. Add comments for edge cases you'd handle -- "TODO: handle rollback of nested transactions" 5. Submit what you have -- partial implementations score points

A stage 4 solution that handles 70% of cases and has clear comments about the remaining 30% scores better than no submission.

---

Anthropic-Specific Notes

Anthropic uses CodeSignal for engineering roles. Based on public information and candidate reports:

• Problems tend toward real-world systems (key-value stores, text processors, scheduling systems) rather than algorithm puzzles
• Python is the most common language choice; TypeScript is also well-supported
• The incremental format tests design thinking as much as implementation ability
• Clean code and good abstractions matter -- the evaluator reads your code, not just runs tests
• There is typically a human review of your code in addition to automated test scoring

Preparation priority: 1. Practice the 8 archetypes in problem-archetypes.md under time pressure (70 min, 4 stages) 2. Internalize the Python patterns from python-patterns-senior.md so they're automatic 3. Do 2-3 full timed practice sessions before the real assessment 4. Get comfortable with the CodeSignal IDE -- practice in it at least once before your assessment

Problem Archetypes for Senior Coding Interviews

Consult this file when practicing specific problem types. Each archetype includes the problem statement, clarifying questions you should ask, a skeleton solution, and follow-up extensions interviewers commonly request.

All examples are in Python 3.10+ and assume a 40-minute interview window.

---

1. In-Memory Key-Value Store

Problem Statement

Design and implement an in-memory key-value store that supports get, set, delete, and exists operations. Keys are strings, values can be any type.

Clarifying Questions to Ask

1. "Should get on a missing key raise an exception or return None?" 2. "Do we need to support TTL (time-to-live) for entries?" 3. "Should operations be thread-safe?" 4. "Is there a maximum capacity? What happens when it's exceeded?" 5. "Do we need transaction support (begin/commit/rollback)?"

Skeleton

from dataclasses import dataclass, field
from typing import Any
import time


@dataclass
class Entry:
    value: Any
    expires_at: float | None = None

    @property
    def is_expired(self) -> bool:
        if self.expires_at is None:
            return False
        return time.monotonic() > self.expires_at


class KeyValueStore:
    def __init__(self) -> None:
        self._data: dict[str, Entry] = {}

    def set(self, key: str, value: Any, ttl_seconds: float | None = None) -> None:
        expires_at = None
        if ttl_seconds is not None:
            expires_at = time.monotonic() + ttl_seconds
        self._data[key] = Entry(value=value, expires_at=expires_at)

    def get(self, key: str) -> Any:
        entry = self._data.get(key)
        if entry is None:
            raise KeyError(f"Key not found: {key}")
        if entry.is_expired:
            del self._data[key]
            raise KeyError(f"Key expired: {key}")
        return entry.value

    def delete(self, key: str) -> bool:
        """Returns True if key existed and was deleted."""
        if key in self._data:
            del self._data[key]
            return True
        return False

    def exists(self, key: str) -> bool:
        entry = self._data.get(key)
        if entry is None:
            return False
        if entry.is_expired:
            del self._data[key]
            return False
        return True

Follow-Up Extensions

Extension 1: Transactions

class KeyValueStore:
    def __init__(self) -> None:
        self._data: dict[str, Entry] = {}
        self._transaction_stack: list[dict[str, Entry | None]] = []

    def begin(self) -> None:
        """Start a new transaction. Transactions can nest."""
        self._transaction_stack.append({})

    def commit(self) -> None:
        """Apply current transaction to parent (or main store)."""
        if not self._transaction_stack:
            raise RuntimeError("No active transaction")
        changes = self._transaction_stack.pop()
        if self._transaction_stack:
            # Merge into parent transaction
            self._transaction_stack[-1].update(changes)
        else:
            # Apply to main store
            for key, entry in changes.items():
                if entry is None:
                    self._data.pop(key, None)
                else:
                    self._data[key] = entry

    def rollback(self) -> None:
        """Discard current transaction."""
        if not self._transaction_stack:
            raise RuntimeError("No active transaction")
        self._transaction_stack.pop()

    def set(self, key: str, value: Any, ttl_seconds: float | None = None) -> None:
        expires_at = None
        if ttl_seconds is not None:
            expires_at = time.monotonic() + ttl_seconds
        entry = Entry(value=value, expires_at=expires_at)
        if self._transaction_stack:
            self._transaction_stack[-1][key] = entry
        else:
            self._data[key] = entry

    def get(self, key: str) -> Any:
        # Check transaction stack top-down (most recent first)
        for txn in reversed(self._transaction_stack):
            if key in txn:
                entry = txn[key]
                if entry is None:
                    raise KeyError(f"Key deleted in transaction: {key}")
                if entry.is_expired:
                    raise KeyError(f"Key expired: {key}")
                return entry.value
        # Fall back to main store
        entry = self._data.get(key)
        if entry is None:
            raise KeyError(f"Key not found: {key}")
        if entry.is_expired:
            del self._data[key]
            raise KeyError(f"Key expired: {key}")
        return entry.value

Extension 2: Periodic Cleanup

import threading

class KeyValueStore:
    def __init__(self, cleanup_interval: float = 60.0) -> None:
        self._data: dict[str, Entry] = {}
        self._lock = threading.Lock()
        self._cleanup_interval = cleanup_interval
        self._start_cleanup_thread()

    def _start_cleanup_thread(self) -> None:
        def cleanup_loop():
            while True:
                time.sleep(self._cleanup_interval)
                self._sweep_expired()

        thread = threading.Thread(target=cleanup_loop, daemon=True)
        thread.start()

    def _sweep_expired(self) -> None:
        now = time.monotonic()
        with self._lock:
            expired_keys = [
                k for k, v in self._data.items()
                if v.expires_at is not None and v.expires_at < now
            ]
            for key in expired_keys:
                del self._data[key]

Complexity: get/set/delete/exists are all O(1). Sweep is O(n) but runs on a background thread.

---

2. File System Abstraction

Problem Statement

Implement an in-memory file system that supports mkdir, create_file, read_file, write_file, ls, and find operations.

Clarifying Questions to Ask

1. "Are paths absolute (always start with /) or can they be relative?" 2. "Should mkdir create intermediate directories (like mkdir -p)?" 3. "Can files and directories have the same name at the same level?" 4. "What should ls return for an empty directory?" 5. "Does find support glob patterns or just exact names?"

Skeleton

from dataclasses import dataclass, field
from typing import Iterator
import fnmatch


@dataclass
class FSNode:
    name: str
    is_dir: bool
    content: str = ""
    children: dict[str, "FSNode"] = field(default_factory=dict)


class FileSystem:
    def __init__(self) -> None:
        self._root = FSNode(name="", is_dir=True)

    def _resolve(self, path: str) -> tuple[FSNode, str]:
        """Resolve path to (parent_node, final_name). Raises if parent missing."""
        parts = [p for p in path.strip("/").split("/") if p]
        if not parts:
            return self._root, ""
        node = self._root
        for part in parts[:-1]:
            if part not in node.children or not node.children[part].is_dir:
                raise FileNotFoundError(f"Directory not found: {part}")
            node = node.children[part]
        return node, parts[-1]

    def mkdir(self, path: str, parents: bool = False) -> None:
        parts = [p for p in path.strip("/").split("/") if p]
        node = self._root
        for i, part in enumerate(parts):
            if part not in node.children:
                if not parents and i < len(parts) - 1:
                    raise FileNotFoundError(f"Parent directory missing: {part}")
                node.children[part] = FSNode(name=part, is_dir=True)
            child = node.children[part]
            if not child.is_dir:
                raise FileExistsError(f"Path exists as file: {part}")
            node = child

    def create_file(self, path: str, content: str = "") -> None:
        parent, name = self._resolve(path)
        if name in parent.children:
            raise FileExistsError(f"Already exists: {name}")
        parent.children[name] = FSNode(name=name, is_dir=False, content=content)

    def read_file(self, path: str) -> str:
        parent, name = self._resolve(path)
        if name not in parent.children:
            raise FileNotFoundError(f"File not found: {name}")
        node = parent.children[name]
        if node.is_dir:
            raise IsADirectoryError(f"Is a directory: {name}")
        return node.content

    def write_file(self, path: str, content: str) -> None:
        parent, name = self._resolve(path)
        if name not in parent.children:
            raise FileNotFoundError(f"File not found: {name}")
        node = parent.children[name]
        if node.is_dir:
            raise IsADirectoryError(f"Is a directory: {name}")
        node.content = content

    def ls(self, path: str = "/") -> list[str]:
        if path == "/":
            node = self._root
        else:
            parent, name = self._resolve(path)
            if name and name in parent.children:
                node = parent.children[name]
            else:
                node = self._root
        if not node.is_dir:
            return [node.name]
        return sorted(node.children.keys())

    def find(self, pattern: str, start: str = "/") -> Iterator[str]:
        """Find files/dirs matching a glob pattern."""
        def _walk(node: FSNode, current_path: str) -> Iterator[str]:
            for name, child in node.children.items():
                child_path = f"{current_path}/{name}"
                if fnmatch.fnmatch(name, pattern):
                    yield child_path
                if child.is_dir:
                    yield from _walk(child, child_path)

        start_node = self._root
        if start != "/":
            parent, name = self._resolve(start)
            start_node = parent.children.get(name, self._root)
        yield from _walk(start_node, start.rstrip("/"))

Follow-Up Extensions

Extension 1: File watchers -- Add callback registration for file changes. Use an observer pattern with on_change(path, callback).

Extension 2: Permissions -- Add read/write/execute bits. Check permissions before operations. Introduces the concept of a "current user."

Extension 3: Symbolic links -- Add symlink(target, link_path). Handle cycle detection in _resolve.

---

3. Rate Limiter

Problem Statement

Implement a rate limiter that allows a configurable number of requests per time window. Support allow(client_id) that returns True if the request should be allowed.

Clarifying Questions to Ask

1. "Fixed window or sliding window?" 2. "Is the limit per-client or global?" 3. "What's the expected number of unique clients?" 4. "Should we support different limits for different clients/tiers?" 5. "Thread-safe?"

Skeleton: Sliding Window with Deque

from collections import defaultdict, deque
import time


class SlidingWindowRateLimiter:
    def __init__(self, max_requests: int, window_seconds: float) -> None:
        self._max_requests = max_requests
        self._window_seconds = window_seconds
        self._requests: dict[str, deque[float]] = defaultdict(deque)

    def allow(self, client_id: str) -> bool:
        now = time.monotonic()
        window = self._requests[client_id]

        # Evict expired timestamps
        cutoff = now - self._window_seconds
        while window and window[0] < cutoff:
            window.popleft()

        if len(window) >= self._max_requests:
            return False

        window.append(now)
        return True

    def remaining(self, client_id: str) -> int:
        """How many requests can this client still make in the current window?"""
        now = time.monotonic()
        window = self._requests[client_id]
        cutoff = now - self._window_seconds
        while window and window[0] < cutoff:
            window.popleft()
        return max(0, self._max_requests - len(window))

Follow-Up: Token Bucket

from dataclasses import dataclass


@dataclass
class Bucket:
    tokens: float
    last_refill: float


class TokenBucketRateLimiter:
    def __init__(self, capacity: int, refill_rate: float) -> None:
        """
        capacity: max tokens in bucket
        refill_rate: tokens added per second
        """
        self._capacity = capacity
        self._refill_rate = refill_rate
        self._buckets: dict[str, Bucket] = {}

    def _get_bucket(self, client_id: str) -> Bucket:
        now = time.monotonic()
        if client_id not in self._buckets:
            self._buckets[client_id] = Bucket(tokens=self._capacity, last_refill=now)
        bucket = self._buckets[client_id]
        # Refill based on elapsed time
        elapsed = now - bucket.last_refill
        bucket.tokens = min(self._capacity, bucket.tokens + elapsed * self._refill_rate)
        bucket.last_refill = now
        return bucket

    def allow(self, client_id: str, cost: float = 1.0) -> bool:
        bucket = self._get_bucket(client_id)
        if bucket.tokens >= cost:
            bucket.tokens -= cost
            return True
        return False

When to use which: Sliding window is simpler and works for most interview problems. Token bucket is better when you need burst handling or variable-cost operations. Mention both, implement whichever fits.

---

4. LRU Cache

Problem Statement

Implement a least-recently-used cache with get, put, and configurable max size. When the cache is full, evict the least recently used entry.

Clarifying Questions to Ask

1. "Should get update the recency of an item?" (Yes -- this is what makes it LRU) 2. "Is there a TTL as well, or just capacity-based eviction?" 3. "Should we support a peek that doesn't update recency?" 4. "What types are keys and values?"

Skeleton: Using OrderedDict

from collections import OrderedDict
from typing import TypeVar, Generic

K = TypeVar("K")
V = TypeVar("V")


class LRUCache(Generic[K, V]):
    def __init__(self, capacity: int) -> None:
        if capacity <= 0:
            raise ValueError("Capacity must be positive")
        self._capacity = capacity
        self._data: OrderedDict[K, V] = OrderedDict()

    def get(self, key: K) -> V:
        if key not in self._data:
            raise KeyError(f"Key not found: {key}")
        self._data.move_to_end(key)  # Mark as recently used
        return self._data[key]

    def put(self, key: K, value: V) -> None:
        if key in self._data:
            self._data.move_to_end(key)
            self._data[key] = value
        else:
            if len(self._data) >= self._capacity:
                self._data.popitem(last=False)  # Evict LRU (first item)
            self._data[key] = value

    def peek(self, key: K) -> V:
        """Get without updating recency."""
        if key not in self._data:
            raise KeyError(f"Key not found: {key}")
        return self._data[key]

    @property
    def size(self) -> int:
        return len(self._data)

    def __contains__(self, key: K) -> bool:
        return key in self._data

    def __repr__(self) -> str:
        items = list(self._data.items())
        return f"LRUCache(capacity={self._capacity}, items={items})"

Follow-Up: Manual Doubly-Linked List

If the interviewer asks you to implement without OrderedDict (rare but possible):

@dataclass
class Node(Generic[K, V]):
    key: K
    value: V
    prev: "Node[K, V] | None" = None
    next: "Node[K, V] | None" = None


class LRUCacheManual(Generic[K, V]):
    def __init__(self, capacity: int) -> None:
        self._capacity = capacity
        self._map: dict[K, Node[K, V]] = {}
        # Sentinel nodes simplify edge cases
        self._head: Node = Node(key=None, value=None)  # type: ignore
        self._tail: Node = Node(key=None, value=None)  # type: ignore
        self._head.next = self._tail
        self._tail.prev = self._head

    def _remove(self, node: Node[K, V]) -> None:
        node.prev.next = node.next  # type: ignore
        node.next.prev = node.prev  # type: ignore

    def _add_to_end(self, node: Node[K, V]) -> None:
        node.prev = self._tail.prev
        node.next = self._tail
        self._tail.prev.next = node  # type: ignore
        self._tail.prev = node

    def get(self, key: K) -> V:
        if key not in self._map:
            raise KeyError(key)
        node = self._map[key]
        self._remove(node)
        self._add_to_end(node)
        return node.value

    def put(self, key: K, value: V) -> None:
        if key in self._map:
            self._remove(self._map[key])
            del self._map[key]
        if len(self._map) >= self._capacity:
            lru = self._head.next  # type: ignore
            self._remove(lru)
            del self._map[lru.key]
        node = Node(key=key, value=value)
        self._add_to_end(node)
        self._map[key] = node

Senior signal: Start with OrderedDict. Mention you know the underlying implementation uses a doubly-linked list. Only implement manually if explicitly asked. This shows pragmatism.

---

5. Task Scheduler

Problem Statement

Implement a task scheduler that supports adding tasks with priorities, executing the highest-priority task, and cancelling pending tasks.

Clarifying Questions to Ask

1. "Is lower number = higher priority, or higher number = higher priority?" 2. "Should tasks with the same priority run FIFO?" 3. "Can tasks have dependencies (task B runs only after task A completes)?" 4. "Should we support recurring tasks?" 5. "What happens if we try to cancel an already-running task?"

Skeleton

from dataclasses import dataclass, field
from typing import Callable, Any
import heapq
import itertools


@dataclass(order=True)
class Task:
    priority: int
    sequence: int  # Tiebreaker for FIFO within same priority
    name: str = field(compare=False)
    fn: Callable[[], Any] = field(compare=False)
    cancelled: bool = field(default=False, compare=False)


class TaskScheduler:
    def __init__(self) -> None:
        self._heap: list[Task] = []
        self._counter = itertools.count()
        self._tasks: dict[str, Task] = {}

    def add_task(self, name: str, fn: Callable[[], Any], priority: int = 0) -> None:
        """Lower priority number = higher priority (runs first)."""
        if name in self._tasks and not self._tasks[name].cancelled:
            raise ValueError(f"Task already exists: {name}")
        task = Task(
            priority=priority,
            sequence=next(self._counter),
            name=name,
            fn=fn,
        )
        self._tasks[name] = task
        heapq.heappush(self._heap, task)

    def cancel(self, name: str) -> bool:
        """Mark task as cancelled. Returns True if task was pending."""
        if name in self._tasks:
            task = self._tasks[name]
            if not task.cancelled:
                task.cancelled = True
                return True
        return False

    def run_next(self) -> Any:
        """Execute and remove the highest-priority non-cancelled task."""
        while self._heap:
            task = heapq.heappop(self._heap)
            if not task.cancelled:
                del self._tasks[task.name]
                return task.fn()
        raise RuntimeError("No pending tasks")

    def pending_count(self) -> int:
        return sum(1 for t in self._tasks.values() if not t.cancelled)

Follow-Up: Task Dependencies

from collections import defaultdict


class DependencyScheduler:
    def __init__(self) -> None:
        self._tasks: dict[str, Task] = {}
        self._dependencies: dict[str, set[str]] = defaultdict(set)
        self._dependents: dict[str, set[str]] = defaultdict(set)
        self._ready: list[Task] = []
        self._counter = itertools.count()
        self._completed: set[str] = set()

    def add_task(
        self, name: str, fn: Callable[[], Any],
        priority: int = 0, depends_on: list[str] | None = None,
    ) -> None:
        task = Task(priority=priority, sequence=next(self._counter), name=name, fn=fn)
        self._tasks[name] = task

        deps = set(depends_on or [])
        # Only wait on dependencies that haven't completed yet
        unmet = deps - self._completed
        self._dependencies[name] = unmet

        for dep in unmet:
            self._dependents[dep].add(name)

        if not unmet:
            heapq.heappush(self._ready, task)

    def run_next(self) -> Any:
        while self._ready:
            task = heapq.heappop(self._ready)
            if task.cancelled:
                continue
            result = task.fn()
            self._completed.add(task.name)
            # Unblock dependents
            for dependent_name in self._dependents.get(task.name, set()):
                self._dependencies[dependent_name].discard(task.name)
                if not self._dependencies[dependent_name]:
                    dep_task = self._tasks[dependent_name]
                    if not dep_task.cancelled:
                        heapq.heappush(self._ready, dep_task)
            return result
        raise RuntimeError("No ready tasks (possible cycle or empty)")

Complexity: add_task is O(log n) for heap push. run_next is O(log n) for heap pop plus O(d) for unblocking dependents where d is the number of direct dependents.

---

6. Event/Pub-Sub System

Problem Statement

Implement an event system where components can subscribe to events and publish events. Support typed events and wildcard subscriptions.

Clarifying Questions to Ask

1. "Are events strings, or should they be typed (enum, class)?" 2. "Should subscribers receive events synchronously or asynchronously?" 3. "Do we need wildcard subscriptions (e.g., subscribe to user.*)?" 4. "Can a subscriber unsubscribe?" 5. "What happens if a subscriber raises an exception?"

Skeleton

from collections import defaultdict
from dataclasses import dataclass
from typing import Callable, Any
import fnmatch


@dataclass
class Subscription:
    callback: Callable[[str, Any], None]
    pattern: str
    is_wildcard: bool


class EventBus:
    def __init__(self) -> None:
        self._exact: dict[str, list[Subscription]] = defaultdict(list)
        self._wildcards: list[Subscription] = []
        self._next_id = 0
        self._sub_map: dict[int, Subscription] = {}

    def subscribe(self, pattern: str, callback: Callable[[str, Any], None]) -> int:
        """Subscribe to events matching pattern. Returns subscription ID."""
        sub = Subscription(callback=callback, pattern=pattern, is_wildcard="*" in pattern or "?" in pattern)
        sub_id = self._next_id
        self._next_id += 1
        self._sub_map[sub_id] = sub

        if sub.is_wildcard:
            self._wildcards.append(sub)
        else:
            self._exact[pattern].append(sub)

        return sub_id

    def unsubscribe(self, sub_id: int) -> bool:
        sub = self._sub_map.pop(sub_id, None)
        if sub is None:
            return False
        if sub.is_wildcard:
            self._wildcards.remove(sub)
        else:
            self._exact[sub.pattern].remove(sub)
        return True

    def publish(self, event: str, data: Any = None) -> int:
        """Publish event. Returns number of subscribers notified."""
        notified = 0
        errors: list[tuple[Subscription, Exception]] = []

        # Exact matches
        for sub in self._exact.get(event, []):
            try:
                sub.callback(event, data)
                notified += 1
            except Exception as e:
                errors.append((sub, e))

        # Wildcard matches
        for sub in self._wildcards:
            if fnmatch.fnmatch(event, sub.pattern):
                try:
                    sub.callback(event, data)
                    notified += 1
                except Exception as e:
                    errors.append((sub, e))

        # Don't let one bad subscriber break others, but surface errors
        if errors:
            for sub, err in errors:
                print(f"Subscriber error for pattern '{sub.pattern}': {err}")

        return notified

Follow-Up Extensions

• Event history: Store last N events for replay when new subscribers join
• Once subscriptions: Subscribe to an event once, auto-unsubscribe after first delivery
• Async delivery: Use asyncio.create_task for non-blocking subscriber notification
• Event filtering: Subscribers can provide a predicate in addition to the pattern

---

7. Log Parser/Analyzer

Problem Statement

Implement a log analyzer that can parse structured log lines, filter by criteria, and compute aggregations (count by level, errors per minute, top endpoints).

Clarifying Questions to Ask

1. "What's the log format? JSON, key=value, or free-form?" 2. "Do logs fit in memory, or should we stream?" 3. "What aggregations are needed?" 4. "Should we support time-range queries?"

Skeleton

from dataclasses import dataclass
from collections import Counter, defaultdict
from datetime import datetime
from typing import Iterator, TextIO
import json


@dataclass
class LogEntry:
    timestamp: datetime
    level: str
    message: str
    metadata: dict[str, str]


class LogAnalyzer:
    def __init__(self) -> None:
        self._entries: list[LogEntry] = []

    def parse_line(self, line: str) -> LogEntry:
        """Parse a JSON log line. Adapt for your format."""
        data = json.loads(line.strip())
        return LogEntry(
            timestamp=datetime.fromisoformat(data["timestamp"]),
            level=data.get("level", "INFO").upper(),
            message=data.get("message", ""),
            metadata={k: v for k, v in data.items()
                      if k not in ("timestamp", "level", "message")},
        )

    def ingest(self, source: TextIO) -> int:
        """Stream log lines from a file-like object. Returns count ingested."""
        count = 0
        for line in source:
            line = line.strip()
            if not line:
                continue
            try:
                entry = self.parse_line(line)
                self._entries.append(entry)
                count += 1
            except (json.JSONDecodeError, KeyError):
                continue  # Skip malformed lines
        return count

    def filter(
        self,
        level: str | None = None,
        start: datetime | None = None,
        end: datetime | None = None,
        message_contains: str | None = None,
    ) -> Iterator[LogEntry]:
        """Lazily filter log entries."""
        for entry in self._entries:
            if level and entry.level != level.upper():
                continue
            if start and entry.timestamp < start:
                continue
            if end and entry.timestamp > end:
                continue
            if message_contains and message_contains not in entry.message:
                continue
            yield entry

    def count_by_level(self) -> dict[str, int]:
        return dict(Counter(e.level for e in self._entries))

    def errors_per_minute(self) -> dict[str, int]:
        """Returns a dict of 'YYYY-MM-DD HH:MM' -> error count."""
        buckets: dict[str, int] = defaultdict(int)
        for entry in self._entries:
            if entry.level == "ERROR":
                minute_key = entry.timestamp.strftime("%Y-%m-%d %H:%M")
                buckets[minute_key] += 1
        return dict(sorted(buckets.items()))

    def top_values(self, metadata_key: str, n: int = 10) -> list[tuple[str, int]]:
        """Top N most common values for a metadata field."""
        counter = Counter(
            e.metadata[metadata_key]
            for e in self._entries
            if metadata_key in e.metadata
        )
        return counter.most_common(n)

Follow-Up Extensions

• Streaming mode: Process logs without storing all entries in memory (use generators, running counters)
• Alerting: Trigger callback when error rate exceeds threshold in a rolling window
• Pattern detection: Find repeated error messages using substring clustering

---

8. API Client with Retry Logic

Problem Statement

Implement an HTTP API client wrapper that supports automatic retries with exponential backoff, circuit breaking, and request/response logging.

Clarifying Questions to Ask

1. "Which HTTP errors should trigger retries? Just 5xx, or also 429?" 2. "Should we support different retry strategies per endpoint?" 3. "Do we need idempotency keys for POST requests?" 4. "Should the circuit breaker be per-endpoint or global?"

Skeleton

from dataclasses import dataclass
from enum import Enum
from typing import Any
import time
import random


class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"


@dataclass
class RetryConfig:
    max_attempts: int = 3
    base_delay: float = 0.5
    max_delay: float = 30.0
    retryable_status_codes: frozenset[int] = frozenset({429, 500, 502, 503, 504})


@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    recovery_timeout: float = 30.0
    success_threshold: int = 2


@dataclass
class Response:
    status_code: int
    body: Any
    headers: dict[str, str]


class APIClient:
    def __init__(
        self,
        base_url: str,
        retry_config: RetryConfig | None = None,
        circuit_config: CircuitBreakerConfig | None = None,
    ) -> None:
        self._base_url = base_url.rstrip("/")
        self._retry = retry_config or RetryConfig()
        self._circuit = circuit_config or CircuitBreakerConfig()
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time: float | None = None

    def _check_circuit(self) -> None:
        if self._state == CircuitState.OPEN:
            if self._last_failure_time is None:
                return
            elapsed = time.monotonic() - self._last_failure_time
            if elapsed >= self._circuit.recovery_timeout:
                self._state = CircuitState.HALF_OPEN
                self._success_count = 0
            else:
                raise ConnectionError(
                    f"Circuit breaker OPEN. Retry after {self._circuit.recovery_timeout - elapsed:.1f}s"
                )

    def _record_success(self) -> None:
        if self._state == CircuitState.HALF_OPEN:
            self._success_count += 1
            if self._success_count >= self._circuit.success_threshold:
                self._state = CircuitState.CLOSED
                self._failure_count = 0
        else:
            self._failure_count = 0

    def _record_failure(self) -> None:
        self._failure_count += 1
        self._last_failure_time = time.monotonic()
        if self._failure_count >= self._circuit.failure_threshold:
            self._state = CircuitState.OPEN

    def _calculate_delay(self, attempt: int) -> float:
        """Exponential backoff with full jitter."""
        exponential = min(self._retry.max_delay, self._retry.base_delay * (2 ** attempt))
        return random.uniform(0, exponential)

    def request(self, method: str, path: str, **kwargs: Any) -> Response:
        """Make an HTTP request with retry and circuit breaker logic.

        In a real implementation, this would use httpx or requests.
        For the interview, we demonstrate the retry/circuit logic.
        """
        self._check_circuit()
        url = f"{self._base_url}/{path.lstrip('/')}"
        last_error: Exception | None = None

        for attempt in range(self._retry.max_attempts):
            try:
                # In real code: response = httpx.request(method, url, **kwargs)
                response = self._make_request(method, url, **kwargs)

                if response.status_code in self._retry.retryable_status_codes:
                    self._record_failure()
                    if attempt < self._retry.max_attempts - 1:
                        delay = self._calculate_delay(attempt)
                        # Handle Retry-After header for 429
                        if response.status_code == 429 and "Retry-After" in response.headers:
                            delay = max(delay, float(response.headers["Retry-After"]))
                        time.sleep(delay)
                        continue
                    return response

                self._record_success()
                return response

            except ConnectionError as e:
                self._record_failure()
                last_error = e
                if attempt < self._retry.max_attempts - 1:
                    delay = self._calculate_delay(attempt)
                    time.sleep(delay)
                    continue

        raise last_error or ConnectionError("All retry attempts failed")

    def _make_request(self, method: str, url: str, **kwargs: Any) -> Response:
        """Placeholder for actual HTTP call. Replace with httpx/requests."""
        raise NotImplementedError("Wire up your HTTP library here")

    def get(self, path: str, **kwargs: Any) -> Response:
        return self.request("GET", path, **kwargs)

    def post(self, path: str, **kwargs: Any) -> Response:
        return self.request("POST", path, **kwargs)

Follow-Up Extensions

• Idempotency keys: Auto-generate UUID for POST/PUT requests, store in header
• Request logging: Log method, URL, status, latency for every request
• Per-endpoint config: Different retry/circuit settings based on path pattern
• Async version: Convert to async/await with httpx.AsyncClient

---

General Tips Across All Archetypes

1. Start with the data model. Before writing any methods, define your @dataclass or core data structures. This grounds the rest of the solution.

2. Write the public API first. Method signatures with type hints and docstrings. Then fill in implementations. This shows top-down design.

3. Use `time.monotonic()` for all timing. Never time.time() -- it can go backwards during NTP adjustments.

4. Generators for streaming. Whenever the follow-up mentions "what if the data doesn't fit in memory?", convert to generators with yield.

5. Mention tests you'd write even if you don't write them: "I'd test the edge case where TTL expires exactly at read time" or "I'd test the race condition where two threads cancel the same task."

6. Name your complexity. After completing each method, state its time and space complexity. Don't wait to be asked.

Senior Python Patterns for Coding Interviews

Consult this file when you need to demonstrate Python fluency that signals Staff+ experience. These are not tricks -- they are patterns that experienced Python engineers use instinctively, and their absence signals a candidate who codes Python but doesn't think in Python.

---

1. Dataclasses for Data Modeling

The signal: Using @dataclass instead of raw dicts or tuples for any structured data with more than 2 fields.

Why it matters: Shows you think about data contracts, not just data passing. Interviewers see raw dicts and think "junior scripting."

from dataclasses import dataclass, field
from typing import Any

# Junior: raw dict
entry = {"key": "user:1", "value": "Alice", "ttl": 300, "created_at": time.time()}

# Senior: structured, type-safe, self-documenting
@dataclass
class CacheEntry:
    key: str
    value: Any
    ttl_seconds: float | None = None
    created_at: float = field(default_factory=time.monotonic)

    @property
    def is_expired(self) -> bool:
        if self.ttl_seconds is None:
            return False
        return time.monotonic() - self.created_at > self.ttl_seconds

When to use @dataclass vs NamedTuple

from typing import NamedTuple

# NamedTuple for immutable records
class Point(NamedTuple):
    x: float
    y: float

# Can be used as dict keys (hashable)
distances: dict[Point, float] = {Point(0, 0): 0.0, Point(1, 1): 1.414}

# Can be unpacked
x, y = Point(3, 4)

---

2. Context Managers for Resource Cleanup

The signal: Using with statements or writing custom context managers for any resource that needs cleanup.

from contextlib import contextmanager
from typing import Iterator


# For database connections, file handles, locks, temporary state
@contextmanager
def transaction(store: "KeyValueStore") -> Iterator["KeyValueStore"]:
    """Context manager that auto-rolls back on exception."""
    store.begin()
    try:
        yield store
        store.commit()
    except Exception:
        store.rollback()
        raise


# Usage
with transaction(store) as txn:
    txn.set("balance:alice", 900)
    txn.set("balance:bob", 1100)
    # If anything raises, both changes roll back

Class-Based Context Manager (for Stateful Resources)

class Timer:
    """Measure execution time of a block."""
    def __init__(self, label: str = "") -> None:
        self.label = label
        self.elapsed: float = 0.0

    def __enter__(self) -> "Timer":
        self._start = time.monotonic()
        return self

    def __exit__(self, *exc_info: Any) -> None:
        self.elapsed = time.monotonic() - self._start
        if self.label:
            print(f"{self.label}: {self.elapsed:.3f}s")


# Usage
with Timer("cache_lookup") as t:
    result = cache.get("user:1")
print(f"Took {t.elapsed:.3f}s")

---

3. Generators for Streaming and Lazy Evaluation

The signal: Using yield instead of building a complete list when the data might be large or the consumer might stop early.

from typing import Iterator


def parse_log_stream(filepath: str) -> Iterator[dict]:
    """Process arbitrarily large log files without loading into memory."""
    with open(filepath) as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError:
                # Log and skip malformed lines
                continue


def filter_errors(entries: Iterator[dict]) -> Iterator[dict]:
    """Composable filter -- chains with other generators."""
    for entry in entries:
        if entry.get("level") == "ERROR":
            yield entry


def first_n(entries: Iterator[dict], n: int) -> list[dict]:
    """Take first N entries -- stops reading early."""
    results = []
    for entry in entries:
        results.append(entry)
        if len(results) >= n:
            break
    return results


# Compose: parse -> filter -> take 10 (reads only as much as needed)
errors = filter_errors(parse_log_stream("/var/log/app.log"))
recent_errors = first_n(errors, 10)

Generator Expression vs List Comprehension

# List comprehension: builds entire list in memory
total = sum([entry.size for entry in entries])  # Wastes memory

# Generator expression: streams values one at a time
total = sum(entry.size for entry in entries)  # O(1) memory

Rule: If you're only iterating once and don't need random access, use a generator expression (no brackets).

---

4. Collections Module Mastery

The signal: Reaching for defaultdict, Counter, deque, OrderedDict naturally, without importing them as an afterthought.

defaultdict -- Eliminates Boilerplate Init Checks

from collections import defaultdict

# Junior: manual key existence check
graph = {}
def add_edge(src, dst):
    if src not in graph:
        graph[src] = []
    graph[src].append(dst)

# Senior: defaultdict handles missing keys
graph: defaultdict[str, list[str]] = defaultdict(list)
def add_edge(src: str, dst: str) -> None:
    graph[src].append(dst)

Counter -- Frequency Analysis in One Line

from collections import Counter

# Count anything hashable
words = ["error", "warn", "error", "info", "error"]
counts = Counter(words)
# Counter({"error": 3, "warn": 1, "info": 1})

# Top N
counts.most_common(2)  # [("error", 3), ("warn", 1)]

# Combine counters
total = counter_a + counter_b
difference = counter_a - counter_b  # Only positive counts

deque -- O(1) Operations at Both Ends

from collections import deque

# Sliding window pattern
class SlidingWindow:
    def __init__(self, window_size: int) -> None:
        self._window: deque[float] = deque()
        self._window_size = window_size

    def add(self, value: float) -> None:
        self._window.append(value)
        while len(self._window) > self._window_size:
            self._window.popleft()  # O(1), not O(n) like list.pop(0)

    @property
    def average(self) -> float:
        if not self._window:
            return 0.0
        return sum(self._window) / len(self._window)

# Also great for BFS
def bfs(graph: dict[str, list[str]], start: str) -> list[str]:
    visited = set()
    queue: deque[str] = deque([start])
    order = []
    while queue:
        node = queue.popleft()
        if node in visited:
            continue
        visited.add(node)
        order.append(node)
        queue.extend(graph.get(node, []))
    return order

---

5. Type Hints in Interview Context

The signal: Type hints on public method signatures. Skip them on internal helpers when time is short.

What to Type

# Always type: class __init__, public methods, return types
class Cache:
    def __init__(self, capacity: int) -> None: ...
    def get(self, key: str) -> Any: ...
    def put(self, key: str, value: Any, ttl: float | None = None) -> None: ...

# Skip typing on: private helpers, local variables, lambda bodies
def _cleanup(self):  # Fine without types in an interview
    expired = [k for k, v in self._data.items() if v.is_expired]
    for k in expired:
        del self._data[k]

Modern Python Type Syntax (3.10+)

# Use | instead of Union (3.10+)
def get(self, key: str) -> str | None: ...

# Use list, dict, tuple directly (3.9+)
def items(self) -> list[tuple[str, Any]]: ...

# Use X | None instead of Optional[X]
def find(self, name: str) -> "Node | None": ...

Interview tip: If you're not sure which Python version the environment supports, ask. If they say "any," use modern syntax -- it shows currency.

---

6. Exception Hierarchy Design

The signal: Defining domain-specific exceptions instead of raising bare Exception or ValueError for everything.

class StoreError(Exception):
    """Base error for all store operations."""
    pass

class KeyNotFoundError(StoreError):
    """Raised when a key doesn't exist in the store."""
    def __init__(self, key: str) -> None:
        self.key = key
        super().__init__(f"Key not found: {key}")

class KeyExpiredError(StoreError):
    """Raised when a key exists but has expired."""
    def __init__(self, key: str) -> None:
        self.key = key
        super().__init__(f"Key expired: {key}")

class TransactionError(StoreError):
    """Raised for transaction-related failures."""
    pass

class CapacityExceededError(StoreError):
    """Raised when store is at capacity and no eviction policy is set."""
    def __init__(self, capacity: int) -> None:
        self.capacity = capacity
        super().__init__(f"Store at capacity: {capacity}")

Why this matters in interviews: It shows you think about error handling as part of API design. Callers can catch StoreError for broad handling or specific subclasses for fine-grained control.

---

7. __slots__ for Performance-Critical Classes

The signal: Knowing when and why to use __slots__, and not using it everywhere.

# Without __slots__: each instance has a __dict__ (40+ bytes overhead)
class Point:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

# With __slots__: no __dict__, fixed attribute set, ~40% less memory
class Point:
    __slots__ = ("x", "y")

    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

When to use: Only when you're creating millions of instances (cache entries, graph nodes, event records). Mention it verbally in an interview ("If we had millions of entries, I'd add __slots__ to reduce memory overhead") -- don't actually do it unless the problem demands it.

When NOT to use: Regular classes with <1000 instances. Classes that need dynamic attribute assignment. Classes that use inheritance heavily (slots + inheritance is tricky).

---

8. functools.lru_cache vs Manual Caching

The signal: Using the stdlib cache for pure functions, writing manual caches for stateful/time-dependent data.

from functools import lru_cache

# Good use of lru_cache: pure computation with repeated inputs
@lru_cache(maxsize=256)
def parse_path(path: str) -> list[str]:
    """Split and normalize a filesystem path. Pure function."""
    return [p for p in path.strip("/").split("/") if p]


# BAD use of lru_cache: function with side effects or time-dependency
@lru_cache(maxsize=100)  # DON'T DO THIS
def get_user(user_id: str) -> dict:
    """Fetches from database -- result can change over time."""
    return db.query(f"SELECT * FROM users WHERE id = {user_id}")

Manual cache when: Data expires, cache needs invalidation, cache needs size limits with custom eviction, or values depend on time/external state.

---

9. Protocol Classes for Structural Typing

The signal: Using Protocol for duck-typing interfaces instead of ABC when you want structural subtyping.

from typing import Protocol, runtime_checkable


@runtime_checkable
class Serializable(Protocol):
    def to_dict(self) -> dict: ...
    def from_dict(cls, data: dict) -> "Serializable": ...


class Storable(Protocol):
    """Any object that can be stored in our cache."""
    @property
    def key(self) -> str: ...
    @property
    def size_bytes(self) -> int: ...


# Any class that has these methods/properties works -- no inheritance needed
class User:
    @property
    def key(self) -> str:
        return f"user:{self.id}"

    @property
    def size_bytes(self) -> int:
        return len(json.dumps(self.__dict__))


def store(item: Storable, cache: "Cache") -> None:
    """Works with any object matching the Storable protocol."""
    if item.size_bytes > cache.max_entry_size:
        raise ValueError(f"Item too large: {item.size_bytes}")
    cache.put(item.key, item)

Interview use: Mention Protocol when the interviewer asks "how would you make this extensible?" It shows you know Python's structural typing beyond just ABCs.

---

10. Testing Hooks: Dependency Injection

The signal: Designing classes so they're testable without mocking frameworks.

from typing import Callable
import time


class RateLimiter:
    def __init__(
        self,
        max_requests: int,
        window_seconds: float,
        clock: Callable[[], float] = time.monotonic,  # Injectable clock
    ) -> None:
        self._max = max_requests
        self._window = window_seconds
        self._clock = clock  # Test can inject a fake clock
        self._requests: dict[str, list[float]] = defaultdict(list)

    def allow(self, client_id: str) -> bool:
        now = self._clock()
        # ... use self._clock() instead of time.monotonic() everywhere


# In tests: inject a controllable clock
class FakeClock:
    def __init__(self, start: float = 0.0):
        self._now = start

    def __call__(self) -> float:
        return self._now

    def advance(self, seconds: float) -> None:
        self._now += seconds


def test_rate_limiter_window_expiry():
    clock = FakeClock(start=0.0)
    limiter = RateLimiter(max_requests=2, window_seconds=60.0, clock=clock)

    assert limiter.allow("client_1") is True
    assert limiter.allow("client_1") is True
    assert limiter.allow("client_1") is False  # Over limit

    clock.advance(61.0)  # Move past the window

    assert limiter.allow("client_1") is True  # Window expired, allowed again

Why this matters: Shows you think about testability as part of design, not as an afterthought. Interviewers at Staff+ level expect this.

---

11. Enum for State Machines

The signal: Using Enum or StrEnum for states instead of string literals.

from enum import Enum, auto


class TaskState(Enum):
    PENDING = auto()
    RUNNING = auto()
    COMPLETED = auto()
    FAILED = auto()
    CANCELLED = auto()

    @property
    def is_terminal(self) -> bool:
        return self in (TaskState.COMPLETED, TaskState.FAILED, TaskState.CANCELLED)

    @property
    def can_cancel(self) -> bool:
        return self in (TaskState.PENDING, TaskState.RUNNING)


# Prevents typos, enables IDE completion, makes invalid states unrepresentable
task.state = TaskState.PENDING  # Type-safe
task.state = "pneding"  # Would be silently wrong with strings

---

12. Patterns to Avoid in Interviews

These patterns are technically valid Python but signal inexperience: