# Data Engineer Interview Questions — Python

Source: Codeayan (https://codeayan.com)
Canonical: https://codeayan.com/get-hired/data-engineer/python
Questions: 56
Last updated: 2026-08-21
Licence: free to read. Please cite Codeayan when quoting.

---

## 1. Explain mutable versus immutable types. Why does the distinction matter the moment you pass something into a function?

*Easy · Very Common*

**Short answer.** Immutable objects such as int, str and tuple cannot be changed after creation; lists, dicts and sets can. Python passes references, so a function receiving a mutable object can modify the caller's data in place. Rebinding the parameter name affects nothing outside the function.

Two functions that look almost identical behave completely differently.

```
def add_topping(order, topping):
    order.append(topping)     # mutates caller's list

def replace_order(order, new_items):
    order = new_items         # rebinds local name only

pizza = ["cheese"]
add_topping(pizza, "olives")
print(pizza)                  # ['cheese', 'olives']

replace_order(pizza, ["paneer"])
print(pizza)                  # ['cheese', 'olives']  ← unchanged
```

The parameter name is a local label pointing at the same object the caller passed. `append` reaches through that label and changes the object itself, which the caller sees. Assignment moves the label to a different object and leaves the original alone.

An immutable object cannot be reached through in that way at all, which is why passing an int or a string is always safe. `s.upper()` returns a new string; it does not modify `s`.

The failure this causes in real code is quiet. A helper that normalises a config dictionary and mutates it in place looks fine in isolation, and then a caller that expected its own dictionary untouched starts seeing modified values three functions away. When a function modifies its argument, either name it so that is obvious, or copy the input at the top and return the new object.

One thing worth knowing for the follow-up: immutability is shallow. A tuple cannot be reassigned, but if it holds a list, that list is still mutable.

```
row = ("BLR", ["delayed"])
row[1].append("cancelled")    # works
```

That also means such a tuple is not hashable and cannot be used as a dictionary key.

**Likely follow-ups**

- If Python passes references, why doesn't reassigning the parameter inside the function change the caller's variable?
- Which built-in types are immutable, and is a tuple always safe to share?
- How would you write a function that clearly signals it modifies its argument?

---

## 2. You have to transform a 200 GB file through three steps and write it out. How do you structure that in Python?

*Hard · Very Common*

**Short answer.** Chain generators so each stage consumes the previous one lazily and yields transformed records. Memory stays flat regardless of file size because only one record is in flight at a time. Each stage stays independently testable, and nothing is read until the final consumer pulls.

Write each stage as a generator that takes an iterable and yields:

```
def read_records(path):
    with open(path, encoding="utf-8") as f:
        for line in f:
            yield line.rstrip("n")

def parse(lines):
    for line in lines:
        parts = line.split("t")
        if len(parts) == 6:
            yield dict(zip(FIELDS, parts))

def enrich(records, lookup):
    for rec in records:
        rec["zone"] = lookup.get(rec["substation"], "UNKNOWN")
        yield rec

def run(path, lookup, out_path):
    pipeline = enrich(parse(read_records(path)), lookup)
    with open(out_path, "w") as out:
        for rec in pipeline:
            out.write(serialise(rec) + "n")
```

Building `pipeline` executes nothing. The final `for` loop pulls one record through all three stages, writes it, and pulls the next. Peak memory is one record plus the lookup table, whether the file is 2 MB or 200 GB.

Three things this structure buys you.

Each stage is a plain function over an iterable, so you test it with a small list and no file at all: `list(parse(["atbt..."]))`. That is much easier than testing a monolithic loop.

Stages compose in any order and new ones slot in without touching the others.

And the file handle stays open only while the generator is alive, which the `with` inside `read_records` handles correctly on normal exhaustion.

Two things that undo the benefit. Calling `list()` anywhere in the chain materialises everything and you are back to loading the file. And sorting or grouping across the whole dataset cannot be done lazily; those need either a bounded key space or an external sort.

For stateful stages such as comparing to the previous record, keep the state in a local variable inside the generator. It survives across yields, which is exactly what makes generators suited to this.

**Likely follow-ups**

- One stage needs to look at the previous record — how does that fit a pipeline of generators?
- Where would you put the error handling so one bad record doesn't kill the run?
- How would you test a stage in the middle of this chain?

---

## 3. When would you use a tuple instead of a list? What does the immutability actually buy you?

*Easy · Very Common*

**Short answer.** Use a list for a homogeneous, growing collection and a tuple for a fixed record whose positions mean different things. Immutability buys you hashability, so tuples work as dictionary keys and set members, and it signals to a reader that the structure will not change.

The convention most teams follow is about meaning, not performance. A list holds many of the same thing and its length varies. A tuple holds a fixed set of fields where position carries meaning.

```
sensor_ids = ["T-01", "T-02", "T-03"]      # list: many of one thing
reading = ("T-01", 34.2, "2026-08-19")     # tuple: one record
```

The concrete capability immutability gives you is hashability. A tuple of immutable values can be a dictionary key or a set member; a list cannot.

```
daily_max = {}
daily_max[("T-01", "2026-08-19")] = 41.7   # fine

daily_max[["T-01", "2026-08-19"]] = 41.7
# TypeError: unhashable type: 'list'
```

That single property is why tuples turn up as composite keys in caches, lookup tables and grouping logic all over real codebases.

The performance argument is usually overstated. Tuples are marginally smaller and construct slightly faster, and unless you are creating crores of them in a hot loop, that difference will not be what makes your job slow. Do not lead with it in an interview; lead with intent and hashability.

Once a tuple has more than three fields, positional access becomes a liability. `reading[2]` tells a reader nothing. `NamedTuple` or a frozen dataclass keeps the immutability and adds names:

```
from typing import NamedTuple

class Reading(NamedTuple):
    sensor_id: str
    celsius: float
    taken_on: str
```

Still hashable, still immutable, and `r.celsius` survives someone reordering the fields.

**Likely follow-ups**

- Is a tuple always faster than a list, and does that difference matter in your code?
- What would you use instead of a tuple when the fields have names?
- Can you use a list as a dictionary key, and why not?

---

## 4. How do you read a value out of a dictionary when the key might not be there?

*Easy · Very Common*

**Short answer.** Square brackets raise KeyError on a missing key. `.get(key)` returns None instead, and `.get(key, default)` returns whatever default you supply. Use brackets when a missing key is a bug you want to hear about, and `.get()` when absence is expected and has a sensible fallback.

```
settings = {"retries": 3, "timeout": None}

settings["region"]              # KeyError: 'region'
settings.get("region")          # None
settings.get("region", "apac")  # 'apac'
```

Choosing between them is a decision about whether absence is normal. Reading a required configuration key with `.get()` and a silent default means a typo in the config file produces a working program with wrong behaviour, which is far harder to debug than a `KeyError` on startup. Reading an optional field from a parsed API response with brackets means one missing attribute crashes the whole batch.

The trap in that example is `timeout`. Both a missing key and a key set to `None` make `.get()` return `None`, so you cannot tell them apart:

```
settings.get("timeout")        # None — but the key exists
settings.get("nonsense")       # None — key doesn't exist

"timeout" in settings          # True
```

When the difference matters, test with `in` rather than inferring from the returned value. This bites when `None` is a legitimate stored value meaning “no limit” and your code treats it as “not configured” and substitutes a default.

One more thing people get wrong: `.get()` never inserts. If you want to read-or-create in one step, that is `.setdefault()`.

```
by_city = {}
by_city.setdefault("Pune", []).append("order-118")
```

For accumulating into a dict repeatedly, `collections.defaultdict(list)` reads better and avoids repeating the default on every line.

**Likely follow-ups**

- What's the difference between .get() returning None and the key existing with a value of None?
- When would you reach for defaultdict instead?
- What does .setdefault() do that .get() doesn't?

---

## 5. Set up logging for a pipeline that several people will debug. What do you log, and in what shape?

*Medium · Very Common*

**Short answer.** Configure handlers and format once in the entry point, and get a module-level logger with `getLogger(__name__)` everywhere else. Log structured key-value context rather than prose, include a run identifier on every line, and log counts at each stage boundary so a partial failure is visible.

Configuration belongs in one place, at the entry point:

```
import logging, sys, json

class JsonFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "ts": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
            "run_id": getattr(record, "run_id", None),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
```

Every other module does one thing:

```
log = logging.getLogger(__name__)
```

That gives you a hierarchy matching your package layout, so you can raise `extraction` to DEBUG while leaving everything else at INFO. Modules that configure their own handlers fight each other and produce duplicated lines.

JSON output matters once logs go to an aggregator, because prose has to be parsed with regex and structured fields do not. A `LoggerAdapter` or a filter attaches the run identifier to every record without passing it through every function:

```
log = logging.LoggerAdapter(log, {"run_id": run_id})
```

Tracing one run through a shared log file is impossible without it.

What to log at each stage: the record count in, the count out, the count rejected, and the elapsed time. Those four numbers turn “the job ran” into “the job dropped 4% at the enrichment step”. Log at boundaries, not per record, or a crore-row job produces a log nobody can read.

`log.exception` inside an `except` block captures the traceback automatically. Use `%s` placeholders rather than f-strings so formatting is deferred and aggregators can group by template.

Never log credentials, tokens, or personal data. Logs are copied, shipped and retained far more widely than the database they came from.

**Likely follow-ups**

- How would you tie every log line from one run together?
- Why configure logging in the entry point rather than in each module?
- What would you never put in a log line?

---

## 6. You copied a list of lists and changing the copy also changed the original. Explain what happened and how you'd fix it.

*Medium · Very Common*

**Short answer.** A shallow copy creates a new outer container holding references to the same inner objects, so mutating an inner object is visible through both copies. `copy.deepcopy` recursively copies everything, giving full independence at the cost of time and memory.

```
import copy

template = [["mon", 0], ["tue", 0]]

shallow = copy.copy(template)      # or template[:] or list(template)
shallow[0][1] = 9
print(template)     # [['mon', 9], ['tue', 0]]  ← changed

deep = copy.deepcopy(template)
deep[1][1] = 5
print(template)     # [['mon', 9], ['tue', 0]]  ← safe
```

The shallow copy did create a new outer list. What it did not do is create new inner lists, so `shallow[0]` and `template[0]` are the same object, and mutating through either label is visible through both.

This turns up most often with a default structure reused across iterations. A roster template copied per employee, where every employee ends up sharing one shift list, produces a bug where updating one person’s schedule silently updates everyone’s. It looks like a database problem until you find the copy.

Deep copy is not free. It walks the entire object graph and reconstructs it, which on a large nested structure inside a loop is genuinely slow. Two alternatives worth mentioning. Rebuild the structure explicitly with a comprehension when you know its shape:

```
fresh = [[day, count] for day, count in template]
```

Or avoid the problem by making the inner elements immutable, since a tuple cannot be mutated through a shared reference at all.

Two details for the follow-ups. `deepcopy` handles cyclic references correctly by keeping a memo of objects it has already copied, so it does not recurse forever. And a class can control what happens by defining `__deepcopy__`, which matters for objects holding a database connection or a file handle that should not be duplicated.

**Likely follow-ups**

- What does deepcopy do when the structure contains a cycle?
- Would this problem arise if the inner elements were tuples of strings?
- How expensive is deepcopy on a large nested structure, and what would you do instead?

---

## 7. Explain `*args` and `**kwargs`. What are they doing on the way in, and on the way out?

*Easy · Very Common*

**Short answer.** In a function signature, `*args` collects extra positional arguments into a tuple and `**kwargs` collects extra keyword arguments into a dict. At a call site, the same symbols do the reverse: they unpack a sequence into positional arguments and a mapping into keyword arguments.

Same syntax, two directions. In the definition it packs; at the call it unpacks.

```
def log_event(event, *tags, **fields):
    print(event, tags, fields)

log_event("upload", "batch", "s3", size=940, retries=2)
# upload ('batch', 's3') {'size': 940, 'retries': 2}

payload = {"size": 940, "retries": 2}
log_event("upload", "batch", **payload)   # unpacking
```

The names are convention only; `*a` and `**kw` work identically. Stick with `args` and `kwargs` because everyone reading your code expects them.

Where these earn their place is a wrapper that must forward whatever it receives without knowing the signature it is wrapping. A retry helper, a timing wrapper, a decorator: all of them accept `*args, **kwargs` and pass them straight through, so they work with any function.

```
def with_retry(fn, *args, **kwargs):
    for attempt in range(3):
        try:
            return fn(*args, **kwargs)
        except TimeoutError:
            continue
    raise
```

Two things worth knowing beyond the basics.

A bare `*` forces everything after it to be keyword-only, which is a good way to stop callers passing a run of unlabelled booleans:

```
def export(path, *, overwrite=False, compress=True):
    ...
export("out.csv", overwrite=True)   # positional not allowed
```

And the cost of `**kwargs` is that it destroys the signature. Callers get no autocomplete, typos become silent extra dict entries rather than errors, and type checkers cannot help. Use it where forwarding is genuinely the point, not to avoid deciding what parameters a function takes.

**Likely follow-ups**

- What does a bare `*` in a signature mean?
- If a caller passes the same argument positionally and by keyword, what happens?
- Why does a decorator almost always accept both of these?

---

## 8. What's wrong with writing `def process(records, seen=[])`? Show me what actually happens.

*Medium · Very Common*

**Short answer.** The default is evaluated once, when the function is defined, not on each call. A default list or dict is therefore shared by every call that omits the argument, and mutations accumulate across calls. Use None as the default and create the container inside the function body.

```
def collect(row, batch=[]):
    batch.append(row)
    return batch

collect("r1")    # ['r1']
collect("r2")    # ['r1', 'r2']  ← not ['r2']
collect("r3")    # ['r1', 'r2', 'r3']
```

The list is created once, at `def` time, and stored on the function object itself. You can see it directly:

```
collect.__defaults__      # (['r1', 'r2', 'r3'],)
```

Every call that omits the argument gets that same list, so state leaks between calls that have nothing to do with each other.

The correct form is a sentinel:

```
def collect(row, batch=None):
    if batch is None:
        batch = []
    batch.append(row)
    return batch
```

`is None` rather than `if not batch`, because an empty list passed deliberately by the caller is falsy and would be silently replaced.

In production this rarely announces itself. A parsing helper with `errors={}` accumulates every error from every file processed since the service started, and memory grows steadily while each individual call looks correct. In a long-running worker that is a slow leak nobody attributes to the function.

The same evaluate-once rule catches other defaults. `def report(as_of=datetime.now())` freezes the timestamp at import time, so a service running for a week reports the same date forever. Pass `None` and compute inside.

Immutable defaults are safe, since there is nothing to mutate: `def f(n=0, label="")` behaves as everyone expects.

There is one deliberate use of the behaviour, worth knowing but not worth writing: attaching a mutable default as a cheap per-function cache. `functools.lru_cache` does the same job without surprising the next reader.

**Likely follow-ups**

- Is `def f(x=[])` ever the behaviour you actually want?
- What would you use as a default for a dict parameter?
- Does the same problem apply to a default of datetime.now()?

---

## 9. Sort these records by department and then by joining date, newest first. Why does the order you apply the sorts matter?

*Easy · Very Common*

**Short answer.** Pass a function to `key=` to extract what you want to sort on, and `reverse=True` to flip the order. Python's sort is stable, meaning equal elements keep their existing relative order, so you can sort by a secondary field first and then by the primary field.

```
staff = [
    {"name": "Riya", "dept": "ops", "joined": "2021-04-01"},
    {"name": "Arun", "dept": "eng", "joined": "2023-07-15"},
    {"name": "Meera", "dept": "ops", "joined": "2019-01-09"},
]

by_dept = sorted(staff, key=lambda r: (r["dept"], r["joined"]))
```

A tuple key sorts by the first element and uses the second to break ties, which handles most multi-field cases in one call.

Mixed directions are where stability earns its keep. You cannot put `reverse=True` on just one element of a tuple key, so you sort twice, least significant first:

```
rows = sorted(staff, key=lambda r: r["joined"], reverse=True)
rows = sorted(rows, key=lambda r: r["dept"])
```

Because Python’s sort is stable, the second sort preserves the date ordering within each department. Reverse the order of those two statements and the result is wrong. For numeric fields there is a shortcut: negate the value in the key, so `key=lambda r: (r["dept"], -r["salary"])`.

`operator.itemgetter` is often faster and clearer than a lambda for straightforward field access:

```
from operator import itemgetter
sorted(staff, key=itemgetter("dept", "joined"))
```

Two things to keep straight. `sorted()` returns a new list and works on any iterable; `.sort()` mutates a list in place and returns `None`. Writing `rows = rows.sort()` silently sets `rows` to `None`, and the failure surfaces later as a confusing `TypeError`.

And the key function is called exactly once per element, then the results are compared. So an expensive key such as a date parse costs n calls rather than n log n, which is the reason `key=` replaced the old comparison-function style.

**Likely follow-ups**

- How would you sort ascending on one field and descending on another in a single pass?
- What's the difference between `sorted()` and `.sort()`?
- Why is `key=` preferred over the old `cmp` style?

---

## 10. Write me a comprehension that builds a dict from these records. Then show me a nested one and tell me where you'd stop.

*Easy · Very Common*

**Short answer.** The same syntax builds lists, dicts and sets: brackets give a list, braces with a colon give a dict, braces without give a set. Nested loops read in the same order you would write them as statements. Beyond two loops or one condition, a plain for loop is easier to follow.

```
books = [("Ghachar Ghochar", 2013), ("Em and the Big Hoom", 2012)]

titles  = [t for t, _ in books]                # list
by_year = {t: y for t, y in books}             # dict
years   = {y for _, y in books}                # set
```

Adding a condition filters, and a conditional expression before the `for` transforms:

```
recent = [t for t, y in books if y >= 2013]
tags = ["new" if y >= 2013 else "old" for _, y in books]
```

Those two `if`s sit in different positions and do different jobs. Filtering goes after the `for`; choosing between two values goes before it. Getting them confused is a common syntax error.

Nesting reads in statement order, outer loop first:

```
shelves = [["a1", "a2"], ["b1"]]
flat = [code for shelf in shelves for code in shelf]
```

Read it as `for shelf in shelves:` then `for code in shelf:`, which is exactly the order you would write the loops. People frequently expect the reverse and get an error about an undefined name.

Where to stop is a judgement worth voicing. Two loops with a condition is around the limit before a comprehension stops being readable, and a reader having to decode it costs more than the two extra lines a loop would take. There is no rule here, only the next person reading it.

One memory point. Brackets build the entire list in memory before returning. Parentheses produce a generator expression instead, which yields lazily and is what you want when feeding a large sequence into `sum` or a loop:

```
total = sum(y for _, y in books)     # no intermediate list
```

Comprehensions have their own scope from Python 3 onwards, so the loop variable does not leak out.

**Likely follow-ups**

- In a nested comprehension, which for loop is the outer one?
- What does a comprehension with parentheses instead of brackets give you?
- Does the loop variable leak into the surrounding scope?

---

## 11. What is a generator, and what do you give up by using one instead of returning a list?

*Medium · Very Common*

**Short answer.** A function containing `yield` returns a generator: calling it runs no code, and each iteration executes until the next yield and pauses there, keeping local state. You gain constant memory over huge sequences. You give up length, indexing, and the ability to iterate more than once.

```
def parse_log(path):
    with open(path) as f:
        for line in f:
            if line.startswith("ERROR"):
                yield line.strip()

for entry in parse_log("app.log"):   # one line in memory
    handle(entry)
```

The list version builds every matching line before the caller sees the first one. On a 40 GB log file that is the difference between a job that runs and one that gets killed by the memory limit.

Calling `parse_log("app.log")` executes nothing. It returns a generator object. The body runs only when something iterates it, advancing to the next `yield` and freezing there with all locals intact.

What you give up matters as much as what you gain.

```
gen = parse_log("app.log")
len(gen)          # TypeError
gen[0]            # TypeError
list(gen)         # works, but defeats the purpose

for x in gen: ... # consumes it
for x in gen: ... # yields nothing, silently
```

That last one is the failure that reaches production. A generator passed to two functions works for the first and gives the second an empty sequence, with no error and no warning. The downstream count comes out as zero and someone spends an afternoon on it. If two passes are needed, materialise once with `list()` and accept the memory cost, or call the generator function again.

One subtlety with the file example: the `with` block stays open while the generator is alive, and if the caller abandons it partway, the file is closed when the generator is garbage collected rather than at a predictable point.

`yield from` delegates to another iterable, which flattens nested generators without an explicit loop and correctly forwards values sent into the generator.

**Likely follow-ups**

- What happens if you iterate the same generator twice?
- How would you get the length of what a generator will produce?
- What does `yield from` do that a loop with yield doesn't?

---

## 12. You're building a query with a value that came from a user. Show me how, and what you'd never do.

*Medium · Very Common*

**Short answer.** Pass values as bound parameters so the driver sends the statement and the data separately, and user input can never be parsed as SQL. String formatting into a query is how injection happens, and it also breaks on ordinary data containing an apostrophe.

```
# never
cur.execute(
    f"SELECT * FROM tenants WHERE block = '{block}'"
)

# always
cur.execute(
    "SELECT * FROM tenants WHERE block = %s", (block,)
)
```

With binding, the database compiles the statement first and then supplies the value, so nothing inside `block` can become executable SQL. Injection is the headline risk, and it is not the only one: a tenant named `D'Souza` breaks the interpolated query with a syntax error, so the same bug that lets an attacker in also fails on real data.

Placeholder style is driver-specific and never quoted. psycopg and mysql-connector use `%s`; sqlite3 uses `?`; SQLAlchemy uses named `:param`. Writing `'%s'` with quotes reintroduces the problem you were solving.

Two things parameters cannot do.

**Identifiers.** Table and column names are not values, so binding does not apply. Validate against an allowlist you control, or use the driver’s quoting helper such as `psycopg.sql.Identifier`. Never format a raw string into that position.

**Variable-length lists.** Generate the right number of placeholders and pass a matching tuple, or on PostgreSQL use `WHERE id = ANY(%s)` with a Python list, which handles any length as one parameter.

An ORM protects its parameterised paths and not raw fragments. `session.execute(text(f"... {value}"))` is exactly as exposed. The safety comes from binding, not from the library.

One point specific to pipelines: interpolating a run date into query text is the same anti-pattern, and it also destroys reproducibility, because the query and the data it produced no longer travel together. Keep the SQL in a file with named parameters and pass the date in.

**Likely follow-ups**

- What do you do when the table name itself has to be dynamic?
- How would you pass a list of a thousand IDs into an IN clause?
- Does using an ORM make you safe by itself?

---

## 13. Walk me through try, except, else and finally. What goes in each one, and why not put everything in the try?

*Medium · Very Common*

**Short answer.** `try` holds the code that might fail, `except` handles a specific failure, `else` runs only if no exception occurred, and `finally` runs either way for cleanup. Keeping the try block narrow matters, because a wide one catches errors from lines you never intended to protect.

```
try:
    conn = connect(dsn)
except ConnectionError as exc:
    log.error("could not reach vendor API: %s", exc)
    raise
else:
    payload = conn.fetch_invoices()   # only if connect worked
finally:
    conn.close() if conn else None
```

The `else` block is the part most people never use, and it is what keeps the `try` narrow. Put `fetch_invoices()` inside the `try` and a `ConnectionError` raised deep inside the fetch gets caught by a handler written for the connection step, and your log says the wrong thing about the wrong operation.

`finally` runs whether or not an exception occurred, and whether or not the block returned. That makes it the right place for releasing a lock, closing a handle, or restoring a setting. In most cases a `with` block expresses the same intent more clearly, since the context manager owns the cleanup.

Three things that separate a working answer from a careful one.

**Catch narrowly.** A bare `except:` catches `KeyboardInterrupt` and `SystemExit` as well, so your job cannot be stopped cleanly. `except Exception` is better and still too broad as a default, since a typo raising `NameError` is swallowed and reported as a vendor failure.

**Never swallow silently.** `except Exception: pass` is how a nightly pipeline reports success while writing nothing. If you catch it, either handle it meaningfully or log it and re-raise.

**Preserve the chain.** When you wrap an error in your own exception type, `raise LoadError("vendor feed failed") from exc` keeps the original traceback attached. Without `from`, the underlying cause is harder to find in the logs.

One sharp edge worth knowing: a `return` inside `finally` discards any exception currently propagating, so an error disappears with no trace.

**Likely follow-ups**

- What's wrong with `except Exception` as your only handler?
- If finally contains a return, what happens to an exception in flight?
- When would you use `raise ... from` rather than a bare raise?

---

## 14. What would you actually write tests for in a data pipeline, and what would you skip?

*Medium · Very Common*

**Short answer.** Test the logic that transforms, validates or decides: parsing, business rules, edge cases you have already been burned by. Skip testing library behaviour and trivial getters. `pytest` needs no boilerplate class; a function named `test_*` with a plain `assert` is a test.

```
# test_tariff.py
import pytest
from billing.tariff import slab_charge

def test_charges_lowest_slab_only():
    assert slab_charge(80) == 320.0

def test_crosses_two_slabs():
    assert slab_charge(150) == 725.0

def test_zero_units_is_free():
    assert slab_charge(0) == 0.0

def test_negative_units_rejected():
    with pytest.raises(ValueError):
        slab_charge(-5)
```

Run `pytest` and it discovers these by name. No class, no `setUp`, no `assertEqual`. When a plain `assert` fails, pytest rewrites it to show both sides of the comparison, which is why the standard library’s `unittest` boilerplate is largely unnecessary now.

What earns a test in a pipeline: the transformation functions, the validation rules, the slab and threshold logic, date and timezone handling, and anything with an edge case someone has already tripped over. Boundary values are where bugs live, so 0, exactly at a slab edge, negative, and empty input matter more than another example in the middle of the range.

What does not: that `json.loads` parses JSON, that a getter returns the attribute, or that a third-party client makes an HTTP call. You are testing someone else’s code and it will fail for reasons unrelated to you.

Two disciplines that matter more than volume.

Every bug you fix gets a test reproducing it first. That is the highest-value test you will ever write, because it is a failure that has actually happened.

Keep the tests fast and free of external dependencies. A suite that needs a live database gets skipped exactly when someone is in a hurry, which is when it was most needed. Pure functions taking data and returning data are trivial to test, which is itself an argument for structuring pipelines that way.

Coverage percentage measures which lines ran, not whether the assertions were meaningful. A test with no assertion still counts toward coverage and verifies nothing.

**Likely follow-ups**

- How would you test something that reads from a database?
- What does a test with no assertion tell you?
- Would you chase a coverage number, and why not?

---

## 15. What does `with` actually guarantee, and how would you write a context manager yourself?

*Medium · Very Common*

**Short answer.** `with` guarantees that setup runs before the block and cleanup runs after it, whether the block finishes normally, returns, or raises. A context manager is any object with `__enter__` and `__exit__`; the `__exit__` method receives the exception details if one occurred.

The guarantee is the point. A `try`/`finally` gives you the same thing, and `with` moves the cleanup logic into the object that owns the resource, so every caller gets it right without remembering.

```
class Stopwatch:
    def __init__(self, label):
        self.label = label

    def __enter__(self):
        self.start = time.perf_counter()
        return self          # bound to `as`

    def __exit__(self, exc_type, exc, tb):
        elapsed = time.perf_counter() - self.start
        log.info("%s took %.2fs", self.label, elapsed)
        return False         # do not suppress

with Stopwatch("model export") as sw:
    export_weights()
```

`__exit__` receives three arguments describing any exception that escaped the block, all `None` if the block finished cleanly. Returning a truthy value from `__exit__` suppresses that exception, which is occasionally what you want and much more often a mistake. Return `False`, or nothing at all, unless you deliberately mean to swallow it.

Whatever `__enter__` returns is what `as` binds. Returning `self` is the common choice; returning `None` is fine when the block does not need a handle.

For most cases the generator form is shorter and reads better:

```
from contextlib import contextmanager

@contextmanager
def temp_setting(obj, **overrides):
    saved = {k: getattr(obj, k) for k in overrides}
    for k, v in overrides.items():
        setattr(obj, k, v)
    try:
        yield obj
    finally:
        for k, v in saved.items():
            setattr(obj, k, v)
```

The `try`/`finally` around the `yield` is essential. Without it, an exception in the body propagates out of the `yield` and the restoration never runs, which is exactly the guarantee you were trying to provide.

One gotcha: opening several resources with a single `with` and commas is fine, but if the second one fails during setup, the first is still cleaned up correctly. Nesting manually with plain assignments does not give you that.

**Likely follow-ups**

- If the body raises, does your cleanup still run, and how does the manager know?
- How would you write one so that it suppresses a specific exception?
- What does `contextlib.contextmanager` do to a generator?

---

## 16. Explain `self`, and what the difference is between an attribute set in `__init__` and one set on the class body.

*Easy · Very Common*

**Short answer.** `self` is the instance, passed automatically as the first argument to every method. An attribute assigned in `__init__` belongs to that instance alone; one assigned in the class body is shared by every instance, and Python falls back to it when the instance has no attribute of that name.

```
class Turbine:
    site = "Jaisalmer"      # class attribute, shared
    log = []                # class attribute, shared and mutable

    def __init__(self, tag):
        self.tag = tag      # instance attribute
```

Lookup goes instance first, then class. So `t.site` finds nothing on the instance and falls back to the class, and every turbine reports the same site. That is useful for genuine constants and defaults.

Assignment does not follow the same path. Writing `t.site = "Bhuj"` creates an instance attribute that shadows the class one for that object only; the class attribute is untouched and every other instance still sees Jaisalmer. Reading falls back, writing does not.

The mutable class attribute is the real trap:

```
a, b = Turbine("T1"), Turbine("T2")
a.log.append("fault")
b.log              # ['fault']
```

`a.log.append(...)` never assigns, so it never creates an instance attribute. It reaches through to the shared list and mutates it. Two objects that should be independent now share state, and in a long-running service that list also grows without bound. Anything mutable belongs in `__init__` as `self.log = []`.

`self` is not a keyword, just a convention. Python passes the instance as the first positional argument to any method called on it, so `t.spin()` is `Turbine.spin(t)`. Rename it and the code works and every reviewer will object.

One detail for the follow-up: `__init__` initialises an object that already exists and must return `None`. The object is created by `__new__`, which you rarely touch outside of immutable subclasses and singletons.

**Likely follow-ups**

- What happens when you assign to an attribute that only exists on the class?
- Why is a mutable class attribute a problem?
- What does `__init__` return, and what actually creates the object?

---

## 17. Why does every Python project tell you to create a virtual environment first?

*Easy · Very Common*

**Short answer.** A virtual environment gives each project its own isolated site-packages directory and its own installed versions. Without one, every project shares the system interpreter's packages, so upgrading a library for one project silently changes behaviour in every other project on the machine.

The problem is a single shared namespace. One project needs an older release of a plotting library that a colleague’s project cannot use, and there is exactly one place to install it.

```
python -m venv .venv
source .venv/bin/activate      # Windows: .venvScriptsactivate
pip install -r requirements.txt
```

`venv` is in the standard library from Python 3.3 onwards, so nothing needs installing first. Activating it puts that environment’s `python` and `pip` at the front of your PATH, so installs land inside the project directory rather than in the system interpreter.

Two failures make this concrete.

Installing into the system Python on Linux can break OS tooling that depends on specific package versions. Recent versions of Debian and Ubuntu now refuse a bare `pip install` outside a virtual environment for exactly this reason, and the error message confuses people who have never used one.

The subtler one is drift. Code runs on your machine and fails in the pipeline, because your machine picked up a library version installed eighteen months ago for something else. Nobody can reproduce it and nobody can explain it. A per-project environment plus a pinned requirements file removes the whole category.

Pin what you install. `pip freeze > requirements.txt` captures exact versions including transitive dependencies, which is what you want for reproducing a deployment. A hand-written file listing only your direct dependencies with ranges is better for a library others will install. Tools like Poetry and uv manage both sides with a lock file, and are worth naming if the interviewer asks what you use.

One practical note: keep `.venv/` out of version control and commit the requirements file instead. The environment is derived; the specification is the artefact.

**Likely follow-ups**

- Two projects need different versions of the same library — what does that look like without a venv?
- What's the difference between `pip freeze` and a requirements file you write by hand?
- Where does a container fit in if you already have virtual environments?

---

## 18. Write me a decorator from scratch and explain what the `@` symbol is actually doing.

*Hard · Very Common*

**Short answer.** `@decorator` above a function is shorthand for reassigning the name: the function is passed to the decorator and the name is rebound to whatever comes back. A decorator is any callable that takes a function and returns a replacement, usually a wrapper closing over the original.

The syntax hides an assignment. These two are identical:

```
@audit
def issue_refund(ticket_id, amount):
    ...

def issue_refund(ticket_id, amount):
    ...
issue_refund = audit(issue_refund)
```

That second form is the whole concept. `audit` receives the original function object and returns something else, and the name now points at the replacement.

Writing one:

```
import functools

def audit(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        log.info("calling %s", fn.__name__)
        result = fn(*args, **kwargs)
        log.info("%s returned", fn.__name__)
        return result
    return wrapper
```

Three details carry the weight. `*args, **kwargs` means the wrapper forwards anything, so one decorator works on functions with any signature. `return result` is easy to omit and turns every decorated function into one that returns None, which is a bug that passes import and fails at runtime. And `functools.wraps` copies the original’s `__name__`, `__doc__` and `__wrapped__` onto the wrapper, without which tracebacks, help output and anything doing introspection all report `wrapper`.

A decorator taking arguments needs one more layer, because `@retry(times=3)` calls `retry(times=3)` first and applies the result as the decorator:

```
def retry(times):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for _ in range(times - 1):
                try:
                    return fn(*args, **kwargs)
                except TimeoutError:
                    continue
            return fn(*args, **kwargs)
        return wrapper
    return decorator
```

Three nested functions is the standard shape and the part people fumble under pressure. Practise writing it.

Two things worth saying. Decorators run at import time, when the `def` is executed, so an expensive decorator body costs you on every import. And methods work unchanged: `self` arrives as the first element of `args`, since by then it is an ordinary positional argument.

**Likely follow-ups**

- Your decorated function's `__name__` now says `wrapper` — how do you fix that?
- How would you write a decorator that takes an argument, like `@retry(times=3)`?
- Can you decorate a method the same way, and what happens to `self`?

---

## 19. Streaming record by record is too slow because your database insert has per-call overhead. How do you batch it without losing the memory guarantee?

*Hard · Common*

**Short answer.** Accumulate into a list, flush when it reaches the chunk size, and reset. Memory stays bounded by the chunk rather than the dataset. The detail people miss is flushing the final partial chunk after the loop ends, which otherwise silently drops the tail.

```
from itertools import islice

def chunked(iterable, size):
    it = iter(iterable)
    while True:
        block = list(islice(it, size))
        if not block:
            return
        yield block

for batch in chunked(parsed_records(), 5000):
    cursor.executemany(INSERT_SQL, batch)
    conn.commit()
```

`islice` pulls at most `size` items from the shared iterator and stops. Because `it` is created once outside the loop, each call continues where the last one left off. Creating the iterator inside the loop instead re-reads from the start forever, which is a genuinely nasty bug to diagnose.

Python 3.12 has `itertools.batched`, which does the same thing and yields tuples. On earlier versions the helper above is the standard shape.

The hand-rolled accumulator version has one line people forget:

```
buffer = []
for rec in records:
    buffer.append(rec)
    if len(buffer) >= 5000:
        flush(buffer)
        buffer = []
if buffer:            # ← the tail
    flush(buffer)
```

Without that final `if`, every run silently drops up to 4,999 records. The job succeeds, the counts are slightly short, and nobody notices until a reconciliation months later.

Chunk size is a real trade-off, not a magic number. Too small and per-call overhead dominates; too large and memory grows and a failure loses more work. Measure with a few sizes against your actual payload rather than guessing, and remember that a chunk of 5,000 wide records is very different from 5,000 narrow ones.

Committing per chunk means a failure leaves earlier chunks applied, so the job must be safely rerunnable. Committing once at the end gives all-or-nothing at the cost of a long transaction, which on PostgreSQL blocks vacuum and on any engine holds locks. For large loads, per-chunk commits plus idempotent writes is the usual answer.

**Likely follow-ups**

- How would you pick the chunk size rather than guessing?
- What happens to the last partial chunk if the loop exits early?
- Would you commit per chunk or once at the end?

---

## 20. You need to remove duplicates from a large collection and then check membership repeatedly. Why a set, and is the O(1) lookup claim really true?

*Easy · Common*

**Short answer.** A set stores elements in a hash table, so membership testing is average O(1) regardless of size, while the same test on a list is O(n). Deduplication is a single constructor call. The costs are that sets hold only hashable objects and do not preserve a meaningful order.

The difference shows up as soon as the collection grows.

```
blocked = set(load_blocked_pincodes())    # 90,000 entries

if pincode in blocked:                    # average O(1)
    reject(pincode)
```

With a list, `in` walks the elements one by one until it finds a match, so checking one lakh incoming pincodes against a 90,000-entry list is roughly nine billion comparisons. Against a set it is one lakh hash lookups.

Set operations replace loops you would otherwise write by hand:

```
served = {"560001", "560034", "560076"}
requested = {"560034", "411001"}

requested & served      # {'560034'}   both
requested - served      # {'411001'}   requested but not served
requested | served      # union
requested ^ served      # in one but not both
```

Three caveats worth raising before the interviewer does.

The O(1) is average case. Hash collisions degrade lookups, and in pathological cases where many elements hash to the same bucket, behaviour approaches linear. For ordinary strings and integers this is not something you will meet, but the honest phrasing is “average O(1)”, not “always O(1)”.

Only hashable objects can go in. A set of lists raises `TypeError`; a set of tuples is fine.

Order is not preserved in any way you should rely on. `set(names)` deduplicates and returns the items in hash order, which will surprise anyone expecting the original sequence. If order matters, `list(dict.fromkeys(names))` deduplicates while keeping first occurrence, since dicts preserve insertion order from Python 3.7 onwards.

**Likely follow-ups**

- What happens to the ordering of your data when you pass it through a set?
- How would you deduplicate while keeping the first occurrence of each item?
- Under what circumstances does that O(1) lookup degrade?

---

## 21. The scheduler retried your load job and now the numbers are doubled. How would you have prevented that?

*Hard · Common*

**Short answer.** Make the job produce the same end state however many times it runs. Delete the partition you are about to write and insert it inside one transaction, or upsert on a business key. Never use `datetime.now()` inside the job; take the logical date as a parameter.

Retries are routine. Schedulers retry on failure, operators rerun a bad day, someone backfills a fortnight. A job that is only correct when it runs exactly once is a job that is regularly wrong.

The delete-write pattern covers most batch loads:

```
def load_day(conn, run_date, rows):
    with conn:                       # commits or rolls back
        cur = conn.cursor()
        cur.execute(
            "DELETE FROM fct_meter_reads WHERE read_date = %s",
            (run_date,),
        )
        cur.executemany(INSERT_SQL, rows)
```

Run it fifty times, get the same table. The transaction is what makes it safe rather than merely repeatable: a crash between the delete and the insert would otherwise leave the day missing entirely.

Three things quietly break idempotency even when the write pattern looks right.

Wall-clock time inside the job. `run_date = date.today()` means rerunning yesterday’s failed task processes today instead, so yesterday stays missing forever. Take the date as an argument and never read the clock.

Files written with a generated name. `output_{uuid4()}.parquet` leaves the previous run’s file behind, and the next reader picks up both. Derive the filename from the logical date.

Appending to a running total in a state table rather than recomputing it. The second run adds again.

Concurrency is the case people forget. Two retries of the same partition can interleave the delete and the insert and produce a partially doubled day. A unique constraint on the natural key, an advisory lock, or a scheduler guarantee of single execution all work. Assuming it cannot happen does not.

Where the target is append-only, idempotency comes from a deduplication key on read, or from writing to a new location and atomically swapping a pointer.

**Likely follow-ups**

- What if the target is an append-only store you can't delete from?
- Two retries run at the same time — does your approach still hold?
- How would you make the job resumable rather than restarting from scratch?

---

## 22. Your extractor hits an upstream service that fails intermittently. Write the retry, and tell me what the cap is for.

*Hard · Common*

**Short answer.** Wait longer after each failure, add randomness to the delay, and stop after a fixed number of attempts. The cap on total wait stops a job blocking a worker slot indefinitely, and jitter prevents many clients retrying in lockstep and keeping a struggling service down.

```
import random, time

TRANSIENT = (TimeoutError, ConnectionResetError, BrokenPipeError)

def with_backoff(fn, attempts=5, base=1.0, cap=30.0):
    for n in range(attempts):
        try:
            return fn()
        except TRANSIENT as exc:
            if n == attempts - 1:
                raise
            delay = min(cap, base * (2 ** n))
            delay += random.uniform(0, delay * 0.3)
            log.warning(
                "attempt %s failed (%s), sleeping %.1fs",
                n + 1, exc, delay,
            )
            time.sleep(delay)
```

Delays run roughly 1, 2, 4, 8 seconds, each with up to 30% added randomly. The `min(cap, ...)` matters because unbounded doubling reaches ten minutes by attempt seven, and a worker asleep for ten minutes is a worker not doing anything else.

Jitter is the part people leave out and the part that matters at scale. Without it, twenty extractors that all failed at the same instant retry at the same instant, and the synchronised wave keeps the service down. Spreading them out is the difference between recovery and a sustained outage.

Which exceptions to retry is the judgement being tested. Timeouts, connection resets and 5xx responses are transient. A malformed request, an authentication failure or a missing resource will fail identically on every attempt, and retrying them wastes the budget and hides the real bug. Catching broadly here turns a five-second failure into a two-minute one with the same outcome.

Two things to raise unprompted.

Retrying is safe for reads and needs thought for writes. A request that timed out may have succeeded upstream, so a blind retry can create a duplicate. Either the endpoint accepts an idempotency key, or the retry checks first.

And in-process retry and orchestrator-level retry solve different problems. In-process handles a blip lasting seconds. A two-hour outage should exhaust the attempts, fail the task, and let the scheduler retry in twenty minutes rather than holding a slot open. Layer them; do not duplicate them.

**Likely follow-ups**

- How do you decide which exceptions are worth retrying at all?
- The upstream is down for two hours — what should your job do?
- Where does this belong if the orchestrator already retries the task?

---

## 23. Walk me through slicing. What does `data[::-1]` do, and does slicing give me the original object or a copy?

*Easy · Common*

**Short answer.** `sequence[start:stop:step]` takes from start up to but not including stop. Negative indices count from the end, and a negative step walks backwards, so `[::-1]` reverses. Slicing a list returns a new list, so modifying the slice does not affect the original.

```
lanes = ["A", "B", "C", "D", "E", "F"]

lanes[1:4]      # ['B', 'C', 'D']   stop is exclusive
lanes[:3]       # ['A', 'B', 'C']
lanes[-2:]      # ['E', 'F']        last two
lanes[::2]      # ['A', 'C', 'E']   every second
lanes[::-1]     # ['F', 'E', ...]   reversed
```

The exclusive stop is what makes `lanes[:3] + lanes[3:]` reconstruct the original with no overlap and no gap, which is the reason for the convention.

Slicing produces a new list. That is why `copy = original[:]` is a common idiom for making a copy:

```
toll_lanes = lanes[:]
toll_lanes.append("G")
print(len(lanes))     # 6 — unaffected
```

The copy is shallow, though. The new list is independent, and the objects inside it are the same objects. With a list of lists, mutating an inner list is visible through both.

Slices behave differently from indexes at the boundaries, which is a genuine gotcha:

```
lanes[10]       # IndexError
lanes[3:99]     # ['D', 'E', 'F'] — no error
lanes[8:12]     # [] — no error
```

An out-of-range slice clamps silently. That is convenient in a paging loop and dangerous when it hides an off-by-one, since a wrong index gives you an empty list rather than an exception.

Two practical notes. `[::-1]` reverses a copy while `list.reverse()` reverses in place, so pick by whether you need the original. And on a very large list, slicing copies every element, so taking `big[:-1]` inside a loop is quietly quadratic; `itertools.islice` iterates without copying.

**Likely follow-ups**

- Your slice copy still shares the inner objects — when does that matter?
- Why does an out-of-range slice not raise IndexError when an out-of-range index does?
- How would you slice every third element starting from the end?

---

## 24. One record in ten million fails to parse. Do you catch it or let the job die? How do you decide?

*Hard · Common*

**Short answer.** Catch what is expected and recoverable at the record level; let anything that invalidates the run propagate. The deciding question is whether continuing produces correct output. A single malformed record can be quarantined; a missing column or a failed connection means every subsequent record is wrong.

Decide at two levels, and be explicit about which you are handling.

Record level, where failure is expected and isolated:

```
rejected = []

for lineno, raw in enumerate(source, start=1):
    try:
        yield parse_manifest_line(raw)
    except (ValueError, KeyError) as exc:
        rejected.append((lineno, str(exc), raw))
        if len(rejected) > MAX_REJECTS:
            raise DataQualityError(
                f"{len(rejected)} bad records, aborting"
            )
```

Job level, where failure invalidates everything: a schema mismatch, an unreachable database, a missing credential. Those should propagate. Catching them and continuing produces a job that reports success while writing nothing, which is the worst outcome available.

The threshold is what turns a rule of thumb into a design. Skipping twelve bad records out of a crore is data cleaning. Skipping eight lakh means the upstream format changed and your job is now silently discarding most of the feed. A rejection ceiling converts the second case into a loud failure. Set it as a proportion, not a fixed count, so it scales with the input.

Three practices that go with this.

Quarantine rather than discard. Write rejected records with their line number and the reason to a file someone can inspect, or the failure is invisible even when counted.

Make partial success visible in the exit status and the logs. A job that processed 92% of records and reported success looks identical to a clean run in any dashboard watching exit codes.

Preserve the cause when you wrap. `raise DataQualityError(...) from exc` keeps the original traceback attached, and without it the log shows your exception and nothing about the underlying parse error.

The question interviewers are really probing is whether you have thought about what happens when the job is half right, because that is the case that actually occurs.

**Likely follow-ups**

- What would change your answer if the failure rate went from 0.1% to 8%?
- Where do the rejected records go, and who looks at them?
- How do you stop a partial failure from being reported as a success?

---

## 25. What's the difference between `is` and `==`? And why does comparing small numbers with `is` seem to work?

*Medium · Common*

**Short answer.** `==` asks whether two objects have equal value; `is` asks whether they are the same object in memory. They give the same answer for small integers and short strings because CPython caches those objects, which misleads people into using `is` for value comparison. Use `is` only for None and other singletons.

```
a = 256
b = 256
a == b      # True
a is b      # True

x = 1000
y = 1000
x == y      # True
x is y      # False
```

Nothing changed about the language between those two blocks. CPython pre-creates integer objects from -5 to 256 and reuses them, so both names point at the identical cached object. At 1000 there is no cache, so two separate objects exist with equal value.

Short strings behave similarly through interning, and the rules for when interning happens vary by how the string was created and by CPython version. Both behaviours are implementation details of CPython, not guarantees of Python, and code that depends on them can break on another interpreter or another release.

The rule that follows: use `==` for value, and `is` only where identity is genuinely the question. In practice that means the singletons.

```
if config is None:          # correct
if config == None:          # works, but non-idiomatic
```

`is None` is preferred for a real reason beyond style. `==` calls the left operand’s `__eq__`, and a class can define that method to return anything it likes, including True when compared with None. `is` cannot be overridden, so it always answers the actual question.

The failure this causes in production is a comparison that passes every test on small values and fails on real data. A validation checking `status_code is 200` works throughout development and breaks when someone loads the code path where the value is parsed from JSON. It reads as a mysterious intermittent bug, and static analysis tools now flag it precisely because it is so common.

**Likely follow-ups**

- Which one would you use to test for None, and why that one?
- Two variables both holding the string "batch" — will `is` return True?
- What method does `==` actually call under the hood?

---

## 26. The same job runs in dev, staging and production with different connection details. How do you manage that?

*Medium · Common*

**Short answer.** Layer the sources: defaults in code, overridden by a config file per environment, overridden by environment variables. Load them once into a typed settings object at startup and validate immediately, so a missing or malformed value fails at launch rather than forty minutes into the run.

```
import os, tomllib
from dataclasses import dataclass
from pathlib import Path

@dataclass(frozen=True)
class Settings:
    warehouse_host: str
    batch_size: int = 5000
    max_rejects_pct: float = 1.0

def load_settings(env: str) -> Settings:
    path = Path(f"config/{env}.toml")
    file_cfg = tomllib.loads(path.read_text()) if path.exists() else {}

    return Settings(
        warehouse_host=os.environ.get(
            "WAREHOUSE_HOST", file_cfg.get("warehouse_host", "")
        ),
        batch_size=int(os.environ.get(
            "BATCH_SIZE", file_cfg.get("batch_size", 5000)
        )),
    )
```

`tomllib` is standard library from Python 3.11 and reads TOML without a dependency.

Three properties make this worth the structure.

Precedence is explicit. Environment beats file beats default, which is the order that lets a container override one value without shipping a new config file.

Validation happens at startup. A `Settings` object constructed in the first second either succeeds or raises, so a typo in `batch_size` fails immediately rather than after the extraction has already run. Add an explicit check for anything that must be present:

```
if not settings.warehouse_host:
    raise RuntimeError("WAREHOUSE_HOST is not set")
```

And the object is frozen, so no code halfway through the run mutates a setting and leaves the next reader confused.

Scattering `os.environ["X"]` through the codebase is what this replaces. You cannot see what the job needs without grepping, missing values surface at unpredictable moments, and everything is a string so every call site does its own conversion.

Two boundaries worth naming. Secrets do not go in the config file; they come from the environment or a secrets manager, and the file holds only non-sensitive settings. And a per-run parameter such as the processing date is not configuration; it is an argument, and it belongs in argparse so a backfill can vary it without editing anything.

**Likely follow-ups**

- What happens on startup if a required setting is missing?
- Why not just read os.environ wherever you need a value?
- Where would a per-run parameter like the processing date live in this scheme?

---

## 27. What counts as falsy in Python? And why is `if x:` not the same test as `if x is not None:`?

*Easy · Common*

**Short answer.** Empty containers, zero, empty strings, None and False are all falsy. So `if x:` is false for an empty list and for zero, not only for None. When you mean "the value was not supplied", test `x is not None`, or a legitimate zero gets treated as missing.

The falsy set is short and worth knowing exactly: `None`, `False`, zero of any numeric type, empty sequences and mappings (`""`, `[]`, `()`, `{}`, `set()`), and any object whose class defines `__bool__` or `__len__` to say so. Everything else is truthy.

The bug this creates is specific and common:

```
def apply_discount(order_total, override=None):
    if not override:
        override = DEFAULT_DISCOUNT
    return order_total * (1 - override)

apply_discount(2000, override=0)   # applies DEFAULT, not 0
```

A caller who explicitly asked for zero discount gets the default instead, because `0` is falsy. Nobody notices until a customer is charged a discount that was meant to be waived. The fix is to test for the thing you actually mean:

```
if override is None:
    override = DEFAULT_DISCOUNT
```

The same shape appears with empty strings, where a deliberately blank field is replaced by a placeholder, and with empty lists, where “query returned no rows” is confused with “query was never run”.

That said, `if items:` is idiomatic and correct when you genuinely mean “is this container non-empty”. Prefer it over `if len(items) > 0:`, which says the same thing more loudly. The judgement is about whether zero and empty are meaningful values in your domain.

One related behaviour to have ready. `or` and `and` return one of their operands, not a boolean:

```
name = user_input or "guest"      # 'guest' if user_input is falsy
```

Convenient, and it carries the same zero-and-empty-string trap.

**Likely follow-ups**

- A function returns an empty list on success — how would you write the caller's check?
- What does `or` return, and how is that different from returning True?
- How would you make your own class falsy when it holds no items?

---

## 28. Your Python function runs as a task in Airflow. What changes about how you write it?

*Medium · Common*

**Short answer.** Write it as a plain function taking explicit parameters, with the logical date passed in rather than read from the clock. Keep it idempotent, since the orchestrator will retry it. Avoid module-level work, because the scheduler imports the DAG file constantly.

```
def load_dispatch_feed(logical_date, source_uri, **_):
    rows = extract(source_uri, logical_date)
    load_day(logical_date, rows)
    return len(rows)
```

Four things differ from a script you run by hand.

**The date comes in, never from the clock.** The orchestrator has a logical date representing the interval being processed, which is not today. Airflow passes it into the callable through the task context, or you template it explicitly. Using `date.today()` means a cleared task from Tuesday processes Thursday, and Tuesday stays permanently missing.

**Retries are guaranteed, so idempotency is not optional.** The task will be rerun, possibly several times, and it must converge to the same state.

**Module-level code runs constantly.** The scheduler parses every DAG file on a short cycle, so anything at import time in that file executes over and over. A database connection or a model load at module level in a DAG file is a real production problem. Keep the DAG file to structure, and put the work inside functions imported from a package.

**Data does not travel between tasks through memory.** Each task typically runs in a separate process or pod, so returning a large object relies on XCom, which serialises through the metadata database and is meant for small values. Pass a path or a table name; write the data to shared storage.

Two more points. A task should be a thin wrapper around a function that also runs standalone, so you can test and debug it without the orchestrator. And log the counts the task produced, since the scheduler UI shows success or failure and nothing about whether the run wrote 90,000 rows or four.

**Likely follow-ups**

- How does the function get the date it is supposed to process?
- Why not pass a large DataFrame between two tasks?
- What happens if the same task is running twice at once?

---

## 29. Write a context manager that acquires a lock and always releases it. Show me both ways of doing it.

*Hard · Common*

**Short answer.** Either define a class with `__enter__` and `__exit__`, or decorate a generator with `@contextlib.contextmanager` and put the setup before the yield and the cleanup in a finally. The generator form is shorter; the class form suits managers that carry state or need reuse.

The class form, in full:

```
class FileLock:
    def __init__(self, path):
        self.path = Path(path)
        self.fh = None

    def __enter__(self):
        self.fh = open(self.path, "x")   # fails if it exists
        self.fh.write(str(os.getpid()))
        return self

    def __exit__(self, exc_type, exc, tb):
        self.fh.close()
        self.path.unlink(missing_ok=True)
        return False       # do not suppress
```

`__exit__` receives the three exception details, all `None` on a clean exit. Returning a truthy value swallows the exception, which is occasionally intended and far more often an accident that hides failures. Return `False`, or nothing.

The generator form does the same job in fewer lines:

```
from contextlib import contextmanager

@contextmanager
def file_lock(path):
    path = Path(path)
    fh = open(path, "x")
    try:
        yield path
    finally:
        fh.close()
        path.unlink(missing_ok=True)
```

The `try`/`finally` around the `yield` is not optional. When the body raises, the exception is thrown back in at the `yield`, and without the `finally` the cleanup never runs, which defeats the entire purpose. This is the single most common mistake with this decorator.

Note where the setup sits in each form. Code before the `try` runs during `__enter__`; if it raises, the manager was never entered and `__exit__` never runs, so anything acquired before that point needs its own handling.

Choose the class form when the manager holds state you want to inspect, needs methods, or must be reentrant. Choose the generator form for straightforward acquire-and-release, which is most cases.

For several resources, nest them in one `with` separated by commas, or use `contextlib.ExitStack` when the number is not known until runtime.

**Likely follow-ups**

- What does returning True from `__exit__` do, and when would you want that?
- Why does the generator form need a try/finally around the yield?
- How would you write one that manages several resources at once?

---

## 30. Write a decorator that takes an argument, like `@timed(threshold=5)`. Why does that need an extra layer?

*Hard · Common*

**Short answer.** `@timed(threshold=5)` calls `timed(threshold=5)` first, and whatever that returns is then applied to the function. So you need three layers: the outer function takes the arguments, returns the actual decorator, which returns the wrapper.

The `@` line with parentheses is doing two steps, not one:

```
@timed(threshold=5)
def compact_segments(...):
    ...

# equivalent to
compact_segments = timed(threshold=5)(compact_segments)
```

`timed(threshold=5)` is called first and must return something that accepts a function. That returned thing is the decorator. Hence three nested functions:

```
import functools, time

def timed(threshold=1.0):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                return fn(*args, **kwargs)
            finally:
                elapsed = time.perf_counter() - start
                if elapsed > threshold:
                    log.warning(
                        "%s took %.2fs", fn.__name__, elapsed
                    )
        return wrapper
    return decorator
```

Read the layers by what each one receives. `timed` gets the configuration. `decorator` gets the function being decorated. `wrapper` gets the runtime arguments. Each returns the next level down.

`functools.wraps` goes on the innermost wrapper, applied to `fn`. Without it, the decorated function’s `__name__` becomes `wrapper`, which then makes the log line above useless and breaks anything doing introspection.

The `finally` means the timing is recorded even when the call raises, which for a pipeline stage is usually what you want.

Two things worth saying. `timed(threshold=5)` and `decorator(fn)` both run once, at import time when the `def` is executed. Only `wrapper` runs per call, so expensive setup can go in the outer layers and cost nothing per invocation.

And forgetting the parentheses is a sharp failure. Writing `@timed` without them passes the function itself as `threshold`, and `decorator` never receives it, so the name ends up bound to the inner `decorator` function. Calling it then produces a bewildering error about arguments. Supporting both forms means checking whether the first argument is callable, which is doable and rarely worth the complexity.

**Likely follow-ups**

- How would you make the parentheses optional so both @timed and @timed(threshold=5) work?
- Where does functools.wraps go in this three-layer version?
- What runs at import time and what runs per call?

---

## 31. Python can't find a variable inside a function. Where does it look, and in what order?

*Medium · Common*

**Short answer.** Python resolves a name through four scopes in order: Local, Enclosing function, Global (module level) and Builtins. The first match wins. Assigning to a name anywhere in a function makes it local for that whole function unless you declare it `global` or `nonlocal`.

LEGB is the lookup order, and the part that catches people is not the lookup but the assignment rule.

```
tax_rate = 0.18

def price_with_tax(base):
    print(tax_rate)          # UnboundLocalError
    tax_rate = 0.05
    return base * (1 + tax_rate)
```

The error appears on a line that only reads the variable. Python scans the whole function body at compile time, sees an assignment to `tax_rate`, and marks it local for the entire function, including lines above the assignment. At the moment of the `print` it is local and not yet bound.

`global` says the name refers to the module-level variable, and `nonlocal` says it refers to the nearest enclosing function’s variable:

```
def make_counter():
    count = 0
    def record():
        nonlocal count       # without this: UnboundLocalError
        count += 1
        return count
    return record
```

`count += 1` is an assignment, so the same rule applies inside the inner function. `nonlocal` points it back out one level. `global` would not work here, since `count` is not at module level.

Two things worth saying in an interview. Only functions, classes, modules and comprehensions create scopes. `for`, `if` and `with` do not, so a loop variable is still visible after the loop ends, which surprises people coming from C-family languages.

And `global` in production code is usually a smell. Module-level mutable state makes functions untestable and behaviour order-dependent, and in a multi-threaded worker it becomes a race condition. Pass the value in and return the result instead.

**Likely follow-ups**

- You assign to a name inside a function that also exists at module level — what does Python decide?
- When would you reach for `nonlocal` rather than `global`?
- Does a for loop or an if block create a new scope?

---

## 32. Two modules import each other and one of them fails with a partially initialised module error. Explain what happened and how you'd fix it.

*Hard · Common*

**Short answer.** A circular import happens when module A imports B while B is still importing A. Python caches the partially built module, so the second import finds it existing but incomplete and the name you want is missing. Restructure so the shared code lives in a third module.

Python records a module in `sys.modules` before executing its body, which is what stops infinite recursion and also what causes the confusing error.

```
# ingest/reader.py
from ingest.writer import flush_batch      # line 1

def read_all(path):
    ...

# ingest/writer.py
from ingest.reader import read_all         # line 1

def flush_batch(rows):
    ...
```

Importing `ingest.reader` starts executing its body, hits line 1, and starts `ingest.writer`. That hits its own line 1 and finds `ingest.reader` already in `sys.modules`, so it does not re-execute it. But that cached module has run only up to line 1, so `read_all` does not exist yet:

```
ImportError: cannot import name 'read_all' from partially
initialized module 'ingest.reader' (most likely due to a
circular import)
```

Three fixes, in order of preference.

**Extract the shared piece.** Usually the cycle exists because both modules need something that belongs in neither. Move it to `ingest/records.py` and have both import from there. This is the real fix, and the cycle is telling you the module boundaries are wrong.

**Import inside the function.** By the time the function runs, both modules are fully loaded, so it works. The cost is that the dependency is hidden from anyone reading the imports, and a typo in the module name fails at call time rather than at startup. Acceptable as a targeted patch, not as a habit.

**Import the module rather than the name.** `import ingest.writer` then `ingest.writer.flush_batch(...)` defers attribute lookup to call time, so the partially built module is fine.

Two notes. Use absolute imports inside packages; relative imports like `from .writer import x` are fine within a package but must never appear in a script run directly. And `TYPE_CHECKING` guards let you import purely for type hints without creating a runtime cycle.

**Likely follow-ups**

- Why does moving the import inside the function work, and when is that a bad idea?
- What does `from . import x` do differently from `import x`?
- How would you find every circular import in a codebase you just inherited?

---

## 33. Three of your pipelines copy the same utility module. Turn it into something they can install.

*Hard · Common*

**Short answer.** Create a `pyproject.toml` declaring the package name, version and dependencies, put the code under a `src/` directory, and install it with pip. Consumers then depend on a version rather than copying files, so a fix reaches every pipeline through one release.

The minimum modern layout:

```
pipelines-common/
  pyproject.toml
  src/
    pipelines_common/
      __init__.py
      parsing.py
```

```
[project]
name = "pipelines-common"
version = "0.3.1"
requires-python = ">=3.10"
dependencies = ["requests>=2.31,

**Likely follow-ups**

- What does `pip install -e .` do that a normal install doesn't?
- Where do you pin versions, and where do you leave them loose?
- How would you make a command-line entry point available after install?

---

## 34. Build me a list of functions in a loop, each one multiplying by its loop index. Then tell me why they all do the same thing.

*Hard · Common*

**Short answer.** A closure captures the variable, not the value it had at definition time. Every function built in the loop refers to the same loop variable, which holds its final value once the loop ends. Bind the value with a default argument or `functools.partial` to fix it.

```
multipliers = []
for factor in [2, 5, 10]:
    multipliers.append(lambda x: x * factor)

[m(3) for m in multipliers]     # [30, 30, 30] ← not [6, 15, 30]
```

Nothing is captured at the moment each lambda is created. Each one holds a reference to the enclosing variable `factor`, and looks it up when called. By the time you call them, the loop has finished and `factor` is 10.

This is late binding, and it is the correct behaviour for closures generally. A closure that captured values would be useless for the counter pattern, where the whole point is that the inner function sees updates to the outer variable.

Two ways to bind the value at definition time. A default argument, which is evaluated once when the lambda is defined:

```
for factor in [2, 5, 10]:
    multipliers.append(lambda x, f=factor: x * f)
```

Or `functools.partial`, which stores the argument explicitly:

```
from functools import partial
from operator import mul

multipliers = [partial(mul, f) for f in [2, 5, 10]]
```

Where this bites in real work is callbacks. Registering a handler per queue inside a loop, or building a list of retry functions each bound to a different endpoint, and every one of them ends up pointing at the last item. The symptom is that all your workers process the same partition, which looks like a configuration problem rather than a language one.

A comprehension has exactly the same issue, since it also builds the functions before any of them run. And each closure holds its variables in a cell object, visible as `fn.__closure__`, which is worth knowing if an interviewer pushes on the mechanism.

**Likely follow-ups**

- Does the same problem show up with a list comprehension building lambdas?
- How would `functools.partial` solve this, and is it clearer?
- What does a closure actually hold — the value or the variable?

---

## 35. When would you use `map` and `filter` with a lambda, and when would you write a comprehension instead?

*Easy · Common*

**Short answer.** A comprehension is usually clearer for transforming or filtering a collection, and it reads left to right. `map` and `filter` earn their place when you already have a named function to pass, since `map(int, values)` needs no lambda at all. Avoid `map` with a lambda.

Compare the same operation three ways on a list of ticket prices:

```
prices = ["450", "1200", "300"]

list(map(int, prices))                    # clean
list(map(lambda p: int(p) * 1.05, prices))
[int(p) * 1.05 for p in prices]           # clearer
```

The first is good code. `map` with an existing callable is compact and says exactly what it does. The second is where readability drops: a lambda inside a map means reading right to left through two layers to work out what happens to each item. The comprehension states the transformation first and the source second, which is how people read.

Filtering follows the same pattern:

```
list(filter(lambda p: int(p) > 400, prices))
[p for p in prices if int(p) > 400]
```

Combining both in one comprehension is natural and needs no nesting:

```
[int(p) * 1.05 for p in prices if int(p) > 400]
```

Two limits on lambdas that are worth stating. A lambda holds one expression, so no statements, no assignments, no try blocks. A conditional expression is allowed because it is an expression: `lambda p: "high" if p > 1000 else "low"`.

And a lambda assigned to a name gains nothing over `def`, while losing a useful name in tracebacks. `charge = lambda x: x * 1.05` shows up as `<lambda>` when it raises, and PEP 8 recommends `def` there.

One behavioural difference worth knowing: in Python 3, `map` and `filter` return lazy iterators, not lists. They consume nothing until iterated and are exhausted after one pass, so a `map` object you iterate twice gives you results the first time and nothing the second.

**Likely follow-ups**

- `map` returns a lazy object in Python 3 — when does that matter?
- Can a lambda contain an if statement, or an assignment?
- How would you write a map and filter together as one comprehension?

---

## 36. You need to fetch from 400 endpoints. Would asyncio help, and how would you write it?

*Hard · Common*

**Short answer.** Yes for I/O-bound work. A single event loop interleaves hundreds of network waits in one thread, because `await` yields control while a response is pending. It does nothing for CPU-bound work, and any blocking call inside a coroutine stalls the entire loop.

```
import asyncio, aiohttp

async def fetch_one(session, url, sem):
    async with sem:
        async with session.get(url, timeout=30) as resp:
            resp.raise_for_status()
            return await resp.json()

async def fetch_all(urls):
    sem = asyncio.Semaphore(20)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, u, sem) for u in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

results = asyncio.run(fetch_all(urls))
```

`asyncio.run` creates the loop, runs the coroutine and cleans up; it landed in Python 3.7 and replaced the manual loop management older examples show.

Where the gain comes from: 400 sequential requests at 200 ms each is 80 seconds of mostly waiting. Awaiting them concurrently in one thread turns that into a few seconds, without the memory cost of 400 threads.

Three things decide whether it actually works.

**Everything in the path must be async.** Calling `requests.get` inside a coroutine blocks the event loop, so all 400 tasks queue behind it and you have written slow synchronous code with extra syntax. `time.sleep` does the same; use `asyncio.sleep`. If a library has no async version, push it to a thread with `asyncio.to_thread`.

**Bound the concurrency.** Firing 400 requests simultaneously will exhaust file descriptors or get you rate-limited. The semaphore caps in-flight requests at 20 while still keeping the pipe full.

**Decide what a failure means.** `gather` without `return_exceptions=True` propagates the first exception and you lose the results that succeeded. With it, exceptions come back as elements in the list and you filter them afterwards.

The boundary worth stating in an interview: asyncio helps when the bottleneck is waiting. For CPU-bound work the loop has nothing to interleave, and you need processes instead.

**Likely follow-ups**

- One blocking call sneaks into a coroutine — what happens to the other 399 tasks?
- How would you stop all 400 firing at once?
- What does await actually do to the event loop?

---

## 37. When would you use ThreadPoolExecutor and when ProcessPoolExecutor? What decides it?

*Hard · Common*

**Short answer.** Threads for I/O-bound work, processes for CPU-bound work. The GIL lets only one thread execute Python bytecode at a time, but it is released during I/O, so threads overlap waiting well and give no speedup on computation. Processes sidestep the GIL at the cost of serialising data between them.

Both share an interface, so switching is one word:

```
from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=16) as pool:
    futures = {
        pool.submit(download_shard, s): s for s in shards
    }
    for fut in as_completed(futures):
        shard = futures[fut]
        try:
            path = fut.result()
        except Exception:
            log.exception("shard %s failed", shard)
```

`as_completed` yields futures in finish order rather than submission order, so a slow item does not hold up the ones behind it. Note that an exception inside a worker is stored, not raised, until you call `.result()`. A pool used without ever reading results will swallow every failure silently.

Choosing between them:

Threads
Processes

Good for
network, disk, database waits
parsing, compression, computation

Memory
shared, cheap
separate interpreter each

Data transfer
free
pickled between processes

Failure isolation
poor
a crash kills one worker

The process pool has a constraint threads do not: everything crossing the boundary must be picklable. Lambdas, local functions, open file handles and database connections all fail. The function must be importable at module level, and each worker builds its own connections.

Serialisation cost also decides it. Sending a large object to a process, computing briefly, and sending it back can cost more than doing the work inline. Processes pay off when the compute per item clearly exceeds the transfer.

On the GIL, be precise about version. Through Python 3.12 the description above holds. Python 3.13 ships an optional free-threaded build with the GIL disabled, which changes the calculus for threads on CPU-bound work, but it is opt-in, not the default, and ecosystem support is still maturing. Saying that accurately is worth more than a blanket claim either way.

**Likely follow-ups**

- What has to be true about the function you submit to a process pool?
- How would you get results back as they finish rather than in order?
- Does the GIL argument still hold on Python 3.13?

---

## 38. Your job takes 40 minutes and you think you know why. How would you check rather than guess?

*Hard · Common*

**Short answer.** Run it under `cProfile` and sort by cumulative time to find where the time actually goes. Developer intuition about bottlenecks is wrong often enough that measuring first is the whole discipline. Profile a representative workload, since a small sample hides the behaviour that matters.

```
python -m cProfile -o profile.out -s cumtime run_job.py
```

Then read it:

```
import pstats

st = pstats.Stats("profile.out")
st.sort_stats("cumulative").print_stats(25)
```

Two columns matter and they answer different questions. `tottime` is time spent inside the function itself, excluding calls it makes. `cumtime` includes everything it called. A function with huge `cumtime` and tiny `tottime` is a coordinator, and the real cost is below it. A function with high `tottime` is doing the work itself and is where an optimisation would land.

Look at `ncalls` alongside them. A function taking 40 microseconds is irrelevant until you see it was called 12 lakh times, and that pattern is the most common finding in real profiling: not one slow function, but a cheap one in a loop that did not need to be there.

Profile something representative. A 500-row sample makes fixed startup costs look dominant and hides the quadratic join that only bites at scale. Profile at a size where the problem actually appears.

Three practical points.

`cProfile` adds overhead, particularly on functions called very frequently, so treat the ratios as meaningful and the absolute numbers as inflated.

It measures the calling thread only, so a threaded or async workload needs care; wall-clock time spent waiting shows differently from CPU time.

For line-level detail inside one function, `line_profiler` with its `@profile` decorator tells you which statement is expensive, which `cProfile` cannot.

For production, sampling profilers such as `py-spy` attach to a running process without restarting it or adding meaningful overhead, which is the only practical option when the slowness will not reproduce locally.

Optimise one thing, re-profile, and check the wall-clock time actually moved. Changing several things at once means not knowing which helped.

**Likely follow-ups**

- cumulative time and total time disagree about which function is worst — which do you trust?
- How would you profile something that only misbehaves in production?
- What would you use to find which line inside a function is slow?

---

## 39. What's the difference between an iterable and an iterator? Which methods does each one need?

*Medium · Common*

**Short answer.** An iterable defines `__iter__` and can produce an iterator. An iterator defines both `__iter__` and `__next__`, holds the position, and raises StopIteration when exhausted. A list is an iterable that hands out a fresh iterator each time, which is why you can loop over it repeatedly.

A `for` loop is doing three things behind the scenes: call `iter()` on the object, call `next()` repeatedly, and stop when `StopIteration` is raised.

```
racks = ["R1", "R2"]
it = iter(racks)          # list -> list_iterator
next(it)                  # 'R1'
next(it)                  # 'R2'
next(it)                  # StopIteration
```

The list itself has no position. Each `iter()` call creates a new iterator with its own cursor, which is why two nested loops over the same list work independently.

An iterator returns itself from `__iter__`, which is what lets you pass one to a `for` loop directly. It also means iterating it consumes it. A file object is its own iterator, so a second loop over an already-read file yields nothing at all, with no error. That is the same silent-empty failure a consumed generator produces.

Writing one by hand:

```
class Countdown:
    def __init__(self, start):
        self.n = start

    def __iter__(self):
        return self

    def __next__(self):
        if self.n

**Likely follow-ups**

- Why can you loop over a list twice but not over a file object twice?
- What does `iter()` do when a class only defines `__getitem__`?
- Where does StopIteration go in a for loop?

---

## 40. You need the index alongside each item, and you need to walk two lists together. Show me both, and tell me what zip does when the lists differ in length.

*Easy · Common*

**Short answer.** `enumerate(seq)` yields index-item pairs and takes a `start` argument. `zip(a, b)` yields tuples pairing corresponding elements and stops at the shortest input, silently dropping the tail of the longer one. Pass `strict=True` in Python 3.10 or later to make a length mismatch raise.

```
crops = ["wheat", "gram", "mustard"]

for i, crop in enumerate(crops, start=1):
    print(f"Plot {i}: {crop}")
```

The `start` argument saves the `i + 1` that otherwise appears everywhere in report-numbering code.

Zip pairs elements positionally:

```
plots = ["P1", "P2", "P3"]
yields = [22.5, 19.1]

list(zip(plots, yields))
# [('P1', 22.5), ('P2', 19.1)]  ← P3 vanishes
```

No error, no warning. The third plot is gone from the output, and if this is feeding a report, the total is short by one row and nothing indicates why. It is a genuinely common source of quietly wrong results when two sequences come from different sources and one is missing a record.

Two ways to guard against it. From Python 3.10, `zip(plots, yields, strict=True)` raises `ValueError` on a mismatch, which is what you want whenever equal length is an invariant. Before 3.10, assert the lengths, or use `itertools.zip_longest` when the shorter sequence should be padded:

```
from itertools import zip_longest
list(zip_longest(plots, yields, fillvalue=None))
```

Combining both is idiomatic and reads well:

```
for i, (plot, y) in enumerate(zip(plots, yields), start=1):
    ...
```

Note the parentheses around the inner pair. Without them the unpacking fails.

Two more things worth knowing. `zip(*rows)` transposes, turning rows into columns, and it is the standard way to unzip a list of pairs back into two tuples. And in Python 3, `zip` and `enumerate` return lazy iterators, so wrap in `list()` if you need to look at the result more than once.

**Likely follow-ups**

- How would you make zip raise instead of truncating?
- What does `zip(*rows)` do?
- Can you unzip back into separate sequences, and what do you get?

---

## 41. Your long-running service starts failing with "too many connections". What went wrong, and how do you fix it?

*Hard · Common*

**Short answer.** Connections are being opened and not returned, usually because an exception skipped the close. Use a pool that reuses a fixed set of connections and a context manager that returns them whatever happens, so the count is bounded by the pool size rather than by request volume.

The leak looks harmless:

```
def fetch_meters(feeder_id):
    conn = psycopg.connect(DSN)
    cur = conn.cursor()
    cur.execute(SQL, (feeder_id,))
    rows = cur.fetchall()
    conn.close()          # skipped when execute raises
    return rows
```

Every exception leaves a connection open. The server holds it until a timeout that may be hours away, and after a few hundred failures the database refuses new connections for everyone, not just this service.

Opening per call is also expensive even when it works. A TCP handshake plus authentication is tens of milliseconds, which dwarfs a fast query.

A pool solves both:

```
from psycopg_pool import ConnectionPool

pool = ConnectionPool(DSN, min_size=2, max_size=10)

def fetch_meters(feeder_id):
    with pool.connection() as conn:      # returned on exit
        with conn.cursor() as cur:
            cur.execute(SQL, (feeder_id,))
            return cur.fetchall()
```

The `with` guarantees the connection goes back to the pool whether the block succeeds, returns or raises. That guarantee is the entire fix; the reuse is a bonus.

Three operational points.

Size the pool against what the database can take, not what your application wants. Total connections across every service and every replica must fit the server’s limit, and a pool of 50 in each of eight workers is 400 connections from one service.

A connection returned mid-transaction poisons the next user. Most pools roll back on return, but check, and never leave a `BEGIN` uncommitted.

Do not create the pool before forking. Child processes inherit the same sockets and corrupt each other’s traffic. Create pools after the fork, or per worker, which matters for Gunicorn and multiprocessing alike.

For short-lived batch jobs, a pool adds little; one connection held for the run is fine. Pools earn their place in anything long-running or concurrent.

**Likely follow-ups**

- What happens to a pooled connection if the previous user left a transaction open?
- How would you size the pool?
- What breaks if you fork the process after creating the pool?

---

## 42. Would you use SQLAlchemy Core or the ORM in a data pipeline? What's the difference?

*Medium · Common*

**Short answer.** Core gives you connection pooling, transactions and a SQL expression language without object mapping, so rows come back as tuples. The ORM adds mapped classes, identity tracking and a unit of work. Pipelines usually want Core, because materialising crores of rows as objects costs memory and time for nothing.

The layers stack. Core sits on the driver and handles pooling, transactions and SQL construction. The ORM sits on Core and maps rows to instances of your classes.

```
from sqlalchemy import create_engine, text

engine = create_engine(DSN, pool_size=5)

with engine.begin() as conn:          # commits on clean exit
    result = conn.execute(
        text("SELECT feeder_id, units FROM reads "
             "WHERE read_date = :d"),
        {"d": run_date},
    )
    for feeder_id, units in result:
        ...
```

Named parameters with `text()` keep you bound rather than interpolating, and `engine.begin()` gives you a transaction that commits at the end of the block or rolls back on an exception.

Why Core suits a pipeline. Bulk work is set-based: insert two lakh rows, aggregate a table, run a merge. The ORM’s value is tracking individual objects and flushing changes, and none of that helps here. Loading two lakh rows as mapped instances allocates two lakh objects, populates identity-map bookkeeping for each, and hands you something you were going to write straight back out.

The classic ORM failure in this context is N+1. Iterating a collection of parent objects and touching a related attribute issues one query per parent, so a job that should be one join becomes fifty thousand round trips. It works on a development database with a hundred rows and collapses in production. `selectinload` or a join fixes it; noticing it needs `echo=True` or query logging.

Where the ORM does earn its place: an application layer with row-level create-and-update logic and validation, and migrations through Alembic, which is worth having regardless of which layer you query through.

Mixing them is normal. Use Core for the extraction and load, and the ORM for the small amount of metadata management around a run.

**Likely follow-ups**

- What is the N+1 problem and how would you spot it?
- Where does the Session fit in, and why does Core not need one?
- How would you see the SQL SQLAlchemy is actually sending?

---

## 43. You're inserting three lakh rows and it takes twenty minutes. What are you doing wrong?

*Medium · Common*

**Short answer.** Almost certainly one insert per row, each with its own round trip and commit. Batch the rows with `executemany` inside a single transaction, or use the database's bulk load path such as PostgreSQL's COPY, which skips statement parsing entirely.

The slow version usually looks reasonable:

```
for row in rows:
    cur.execute(INSERT_SQL, row)
    conn.commit()          # ← per row
```

Two costs stack up. Each `execute` is a network round trip, and at 2 ms that alone is ten minutes for three lakh rows. Each `commit` forces a disk flush, which is the larger of the two.

Batching fixes both:

```
with conn:                              # one transaction
    cur.executemany(INSERT_SQL, rows)
```

`executemany` sends parameter sets together rather than one statement per row. How much it actually helps depends on the driver: psycopg 3 and recent MySQL connectors genuinely pipeline these, while some older drivers loop internally and give you little beyond the single commit. Check yours rather than assuming.

For PostgreSQL, `COPY` is a different order of magnitude because it bypasses statement parsing and planning altogether:

```
with cur.copy(
    "COPY toll_transits (plaza_id, vehicle_class, amount) "
    "FROM STDIN"
) as copy:
    for row in rows:
        copy.write_row(row)
```

Combine batching with the chunking discipline you would apply anyway, committing every few thousand rows, so a failure does not lose the whole run and the transaction does not grow unbounded.

Two things worth raising. A batch fails as a unit, so one malformed row in five thousand rejects all of them. Either validate before the batch or catch the failure and fall back to row-by-row for that chunk only, to isolate the culprit.

And on a large load, dropping non-essential indexes before inserting and rebuilding afterwards is often the biggest single win, since every index must be updated per row.

**Likely follow-ups**

- How would you handle one bad row inside a batch of five thousand?
- When would COPY beat executemany, and by how much?
- Where does autocommit fit into this?

---

## 44. What's actually wrong with writing `except:` with nothing after it?

*Medium · Common*

**Short answer.** A bare `except:` catches everything derived from BaseException, including KeyboardInterrupt and SystemExit, so your process cannot be interrupted or shut down cleanly. It also swallows genuine bugs such as typos raising NameError, turning them into whatever your handler pretends went wrong.

```
while True:
    try:
        job = queue.pop()
        proces_job(job)          # typo
    except:
        log.warning("job failed, retrying")
```

Two things are broken here and neither is visible. The typo raises `NameError` on every single iteration, and the handler reports it as a job failure, so the log fills with retry messages about a queue that is working perfectly. And pressing Ctrl-C raises `KeyboardInterrupt`, which the bare except catches, so the loop keeps running and the only way out is to kill the process.

The exception hierarchy is why. `KeyboardInterrupt`, `SystemExit` and `GeneratorExit` inherit from `BaseException` directly, not from `Exception`, precisely so that ordinary handlers do not intercept them. A bare `except:` sits at the `BaseException` level and undoes that design.

```
except Exception:        # excludes Ctrl-C and sys.exit
except (TimeoutError, ConnectionError):   # better still
```

Catch what you can actually do something about. If the handler’s response is “retry the network call”, then catch the network errors, and let a `KeyError` from malformed data propagate to somewhere that knows what to do with it.

There are two legitimate uses of a broad catch. A top-level supervisor loop in a long-running worker, where the job must not die because one item failed, and a plugin boundary where you cannot know what user code raises. In both cases the handler must log the full traceback and re-raise or record the failure:

```
except Exception:
    log.exception("job %s failed", job.id)   # includes traceback
    failed.append(job.id)
```

`log.exception` rather than `log.warning` is the detail that matters. Without the traceback you have a message and no way to find the line.

Interviewers ask this because a swallowed exception is the hardest class of bug to debug: the program reports success while doing nothing.

**Likely follow-ups**

- So what's the narrowest handler you'd write for a network call that might time out?
- If you must catch broadly, what do you do inside the handler?
- What's the difference between `except Exception` and `except BaseException`?

---

## 45. The pipeline currently writes JSON. Someone suggests Parquet. What would you tell them?

*Medium · Common*

**Short answer.** Parquet is columnar and compressed, so it is far smaller on disk and much faster when you read a few columns out of many. JSON is row-oriented, human-readable and schemaless, which suits APIs and small configs. Avro sits between them as a row format with a strict evolving schema.

The three formats answer different questions.

JSON
Parquet
Avro

Layout
row, text
columnar, binary
row, binary

Schema
none
embedded
required, evolvable

Size
largest
smallest
middle

Read a few columns
full scan
reads only those
full scan

Readable by eye
yes
no
no

Writing Parquet from Python:

```
import pyarrow as pa
import pyarrow.parquet as pq

table = pa.Table.from_pylist(records)
pq.write_table(table, "transits.parquet",
               compression="snappy")
```

The size difference is not marginal. A JSON file with repeated keys on every record routinely compresses to a fifth or less as Parquet, because a column of similar values encodes far better than interleaved rows. Storage cost aside, that is less to read.

Column pruning is where the read speed comes from. A table with sixty columns where your query touches four reads four column chunks and skips the rest, which JSON cannot do at all since every field of every record must be parsed to reach the one you want.

Two things to raise before switching.

Parquet is not appendable in any useful sense. You write files, and updating means rewriting a file or a partition. If your pipeline appends records continuously, that shapes the design, and it is why partitioned directories are the norm.

Small files hurt. Thousands of tiny Parquet files carry per-file footer overhead and destroy the scan advantage, so batch writes to a reasonable size.

Avro’s case is schema evolution. It carries a formal schema with defined rules for adding and removing fields, which suits a streaming feed where producers and consumers upgrade independently.

Keep JSON for API payloads, configuration, and anything a human will open. For analytical data at rest, Parquet is usually the answer.

**Likely follow-ups**

- A new field appears in the source next month — which format handles that best?
- What does column pruning actually save you at read time?
- When would you keep JSON despite everything?

---

## 46. You've been handed a 30 GB CSV export and 8 GB of RAM. How do you process it?

*Medium · Common*

**Short answer.** Iterate the file object directly, which yields one line at a time and holds only that line in memory. Avoid `.read()` and `.readlines()`, both of which pull the whole file in. For fixed-size chunks of binary data, read in a loop with an explicit chunk size.

```
# loads 30 GB into RAM, then dies
rows = open("exports/claims.csv").readlines()

# streams, one line at a time
with open("exports/claims.csv") as f:
    header = next(f)
    for line in f:
        process(line)
```

A file object is its own iterator, so the `for` loop reads a buffered chunk from disk and hands you one line per iteration. Memory stays flat regardless of file size. That `next(f)` for the header works because the same iterator is consumed by both.

For CSV specifically, `csv.reader` wraps the same streaming behaviour and handles quoted fields containing newlines, which naive line splitting gets wrong. Reach for it rather than `line.split(",")` on anything that came out of a spreadsheet.

Binary or fixed-size reading needs an explicit loop:

```
with open(path, "rb") as f:
    while chunk := f.read(1024 * 1024):
        digest.update(chunk)
```

The walrus operator needs Python 3.8 or later; before that it is a `while True` with a break.

Two things that decide whether streaming actually works for your problem.

Aggregates are fine if they are incremental. Running totals, counts, min and max all work in one pass with constant memory. A median or an exact distinct count needs the whole dataset, so either accept the memory, spill to a temporary file, or use an approximate algorithm.

The failure mode people hit anyway is accumulating inside the loop. Streaming the file and appending every parsed row to a list uses the same memory as reading it all at once, just more slowly. Write results out as you go, or aggregate into something bounded.

If the work per line is genuinely CPU-heavy, `multiprocessing` over line ranges helps; if it is I/O bound, it mostly will not, since one disk is the constraint.

**Likely follow-ups**

- What if you need a total that depends on the whole file?
- How would you handle a record that spans multiple lines?
- Where would you parallelise this, and what stops you?

---

## 47. Your function calls a vendor API. How do you test it without hitting the vendor?

*Hard · Common*

**Short answer.** Replace the external call with a controlled stand-in using `unittest.mock.patch`, and patch the name where it is looked up, not where it is defined. Set the return value or side effect you need, then assert both on the result and on how the mock was called.

```
# shipping/rates.py
import requests

def quote_freight(pincode, weight_kg):
    resp = requests.get(RATE_URL, params={...}, timeout=10)
    resp.raise_for_status()
    return resp.json()["amount"] * weight_kg
```

```
from unittest.mock import patch, Mock

@patch("shipping.rates.requests.get")
def test_quote_multiplies_by_weight(mock_get):
    mock_get.return_value = Mock(
        status_code=200,
        json=lambda: {"amount": 42.5},
    )
    assert quote_freight("560001", 4) == 170.0
    assert mock_get.call_count == 1
```

The patch target is the part people get wrong, and it is worth stating clearly. You patch `shipping.rates.requests.get`, not `requests.get`. The module under test looked up the name at import time and holds its own reference, so patching the original location leaves that reference untouched and the real request goes out. The rule is: patch where it is used, not where it is defined.

`side_effect` covers the cases `return_value` cannot:

```
mock_get.side_effect = [
    requests.Timeout(),
    Mock(status_code=200, json=lambda: {"amount": 42.5}),
]
```

Successive calls get successive items, so this tests that a retry recovers after one failure. An exception class as `side_effect` raises it every time.

Assert on the call, not only the result. `mock_get.assert_called_once_with(...)` catches a change that sends the wrong parameters while still returning something plausible.

Two boundaries worth naming. Over-mocking produces tests that pass while the system is broken, because you have asserted your assumptions about the vendor rather than reality. And mocking the thing you are testing rather than its dependency tests nothing at all. For HTTP specifically, `responses` or `requests-mock` intercept at the transport layer and are less brittle than patching function names.

**Likely follow-ups**

- You patched the wrong path and the real call still went out — what did you get wrong?
- How would you assert that the retry actually happened three times?
- When is mocking the wrong answer entirely?

---

## 48. Ten tests all need the same sample records and a temp directory. How do you set that up in pytest?

*Medium · Common*

**Short answer.** Write a fixture: a function decorated with `@pytest.fixture` that returns or yields the thing tests need. Any test that names it as a parameter receives it. Fixtures compose, clean up after themselves with a `yield`, and have a scope controlling how often they run.

```
import pytest

@pytest.fixture
def sample_dispatches():
    return [
        {"awb": "A1", "weight": 2.4, "zone": "N"},
        {"awb": "A2", "weight": 0.0, "zone": "S"},
    ]

@pytest.fixture
def staging_dir(tmp_path):
    d = tmp_path / "staging"
    d.mkdir()
    return d

def test_rejects_zero_weight(sample_dispatches):
    clean, bad = validate(sample_dispatches)
    assert len(bad) == 1
```

The parameter name is the wiring. Pytest matches it against fixtures in the file, then in `conftest.py`, then built-ins. `tmp_path` is one of those built-ins, giving a fresh temporary directory per test that pytest cleans up, and `staging_dir` composes on top of it just by taking it as a parameter.

For setup that needs teardown, `yield` splits the fixture:

```
@pytest.fixture
def warehouse_conn():
    conn = connect(TEST_DSN)
    yield conn
    conn.rollback()
    conn.close()
```

Everything after the `yield` runs once the test finishes, pass or fail.

Scope controls how often it runs. The default is per test, which is the safe choice. `scope="session"` runs it once for the entire suite, which suits an expensive read-only resource such as a test container, and is a trap for anything mutable: one test appends to the shared list and the next test sees it, so your suite passes or fails depending on the order it ran in. That class of bug is genuinely hard to track down.

Put shared fixtures in `conftest.py` at the appropriate directory level. Pytest discovers them automatically with no import, and files deeper in the tree inherit from the ones above.

`@pytest.mark.parametrize` is the companion feature: one test function, many input and expected pairs, each reported separately.

**Likely follow-ups**

- Two tests mutate the same fixture and one of them starts failing — why?
- What does scope="session" change, and what does it risk?
- Where does a fixture go so several test files can use it?

---

## 49. Turn this script into a proper command-line tool with subcommands. argparse or click?

*Medium · Common*

**Short answer.** `argparse` is standard library and does everything most tools need, including subcommands. `click` is a dependency that removes boilerplate through decorators and handles nesting and prompting more cleanly. For an internal tool with a handful of commands, either is defensible.

argparse with subcommands:

```
import argparse

def build_parser():
    p = argparse.ArgumentParser(prog="soil")
    p.add_argument("-v", "--verbose", action="store_true")
    sub = p.add_subparsers(dest="command", required=True)

    ing = sub.add_parser("ingest", help="load a survey file")
    ing.add_argument("path")
    ing.set_defaults(func=cmd_ingest)

    rep = sub.add_parser("report", help="build the summary")
    rep.add_argument("--district", required=True)
    rep.set_defaults(func=cmd_report)
    return p

def main(argv=None):
    args = build_parser().parse_args(argv)
    configure_logging(args.verbose)
    return args.func(args)
```

`set_defaults(func=...)` is the idiom that keeps this tidy: each subparser carries its handler, so `main` dispatches with one call instead of a chain of comparisons.

The same thing in click is shorter, because the decorators carry the metadata:

```
import click

@click.group()
@click.option("--verbose", is_flag=True)
def cli(verbose):
    configure_logging(verbose)

@cli.command()
@click.argument("path", type=click.Path(exists=True))
def ingest(path):
    ...
```

`click.Path(exists=True)` validating the file before your code runs is representative of the difference: argparse can do it with a custom `type=` function, click gives it to you.

Three points that matter regardless of library.

`main(argv=None)` taking an argument list makes the CLI testable, because a test can call `main(["ingest", "x.csv"])` without touching `sys.argv`.

Return an exit code and let the entry point pass it to `sys.exit`. A tool that fails and exits zero tells the scheduler everything is fine, which is worse than crashing.

Wire the verbosity flag into logging configuration at the entry point, not into scattered print statements.

**Likely follow-ups**

- How would you make the tool installable so it runs by name from anywhere?
- Where would a `--verbose` flag connect to your logging setup?
- What exit code should the tool return when the job fails?

---

## 50. What's the difference between `__str__` and `__repr__`? Which one shows up when you print a list of your objects?

*Medium · Common*

**Short answer.** `__repr__` is for developers and should be unambiguous; `__str__` is for end users and should be readable. Printing a container calls `__repr__` on each element, not `__str__`, so define `__repr__` first. `__str__` falls back to `__repr__` when it is missing.

```
class Route:
    def __init__(self, code, stops):
        self.code = code
        self.stops = stops

    def __repr__(self):
        return f"Route({self.code!r}, stops={self.stops})"

    def __str__(self):
        return f"{self.code} ({self.stops} stops)"

r = Route("KA-17", 12)
print(r)         # KA-17 (12 stops)
print([r])       # [Route('KA-17', stops=12)]
r                # Route('KA-17', stops=12)  in the REPL
```

The list case is the one people get wrong. Containers always call `__repr__` on their elements, so a class with only `__str__` defined prints beautifully on its own and shows `<__main__.Route object at 0x7f...>` inside a list, in a debugger, and in a log line that formats a collection. If you write only one of the two, write `__repr__`.

The `!r` in the f-string calls `repr()` on the field, which keeps quotes around strings so the output stays unambiguous.

`__eq__` changes comparison from identity to whatever you define:

```
def __eq__(self, other):
        if not isinstance(other, Route):
            return NotImplemented
        return self.code == other.code
```

Returning `NotImplemented` for unknown types lets Python try the reflected operation instead of forcing a wrong answer.

The consequence people forget: defining `__eq__` sets `__hash__` to None, so instances become unhashable and cannot go in a set or be used as dict keys. If you need both, define `__hash__` over the same fields, and only when those fields are immutable. A hashable object whose hash changes after insertion becomes unfindable in the dict it is sitting in.

`@dataclass` generates `__init__`, `__repr__` and `__eq__` from annotations, and `frozen=True` adds a `__hash__`. For plain data-carrying classes it is usually the better starting point.

**Likely follow-ups**

- If you define `__eq__`, what else do you have to define and why?
- What would you use `__hash__` for after that?
- How does a dataclass change this whole picture?

---

## 51. Why does everyone write `if __name__ == "__main__":` at the bottom of a script?

*Easy · Common*

**Short answer.** Python sets `__name__` to `"__main__"` in the file being run directly, and to the module's name when it is imported. The guard runs your script logic only in the first case, so importing the module for its functions does not execute the whole job as a side effect.

```
# ingest.py
def load_readings(path):
    ...

def main():
    load_readings("data/august.csv")
    publish_summary()

if __name__ == "__main__":
    main()
```

Run `python ingest.py` and `__name__` is `"__main__"`, so `main()` fires. Write `from ingest import load_readings` in a notebook or another module, and `__name__` is `"ingest"`, so the guard is false and only the definitions are created.

Without it, every import runs the entire job. Someone imports one helper function and the full August ingestion kicks off, writing to the production summary table. It is not a subtle failure, and it is one people meet exactly once.

Testing is the everyday version of the same problem. A test module importing your script to test one function triggers the whole pipeline before any test runs, so the suite is slow and has side effects nobody asked for.

Three related points.

Anything at module level runs on import, guard or not. A database connection or an expensive model load written at the top of the file executes for every importer, so those belong inside `main()` or a function that is called deliberately.

`multiprocessing` on Windows and on macOS with the spawn start method re-imports the main module in each child process. Without the guard, each child runs your script again, which spawns more children, and the process count grows until something gives.

Keep `main()` thin and the guard thinner. Argument parsing goes inside `main()` rather than at module level, so importing the module does not attempt to read `sys.argv` from whatever process happens to be running.

**Likely follow-ups**

- What breaks specifically when multiprocessing imports your module on Windows?
- Is the guard needed in a file that will only ever be imported?
- Where would you put the argument parsing relative to the guard?

---

## 52. Containerise this Python job. What goes in the Dockerfile and what commonly goes wrong?

*Medium · Common*

**Short answer.** Start from a pinned Python base image, install dependencies as a separate layer before copying the code so the layer caches, and run as a non-root user. Set `PYTHONUNBUFFERED=1` or your logs will not appear until the process exits.

```
FROM python:3.12-slim

ENV PYTHONUNBUFFERED=1
    PYTHONDONTWRITEBYTECODE=1

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY src/ ./src/

RUN useradd --create-home runner
USER runner

ENTRYPOINT ["python", "-m", "src.jobs.recon"]
```

The ordering is the important part. Docker caches layers, and a layer is invalidated when its inputs change. Copying `requirements.txt` alone and installing before copying the source means a code change reuses the cached install layer. Copy everything first and every one-line edit reinstalls every dependency, turning a ten-second build into three minutes.

`PYTHONUNBUFFERED=1` is the one people discover the hard way. Without it, stdout is block-buffered when not attached to a terminal, so a job’s log lines sit in a buffer and appear all at once at exit, or not at all if the container is killed. Debugging a job that appears to produce no output until it dies is a miserable afternoon.

Three more points.

Pin the base image tag. `python:3.12-slim` is better than `python:latest`, which changes under you and eventually breaks a build with no code change. Pinning the digest is stricter still.

Slim images lack a compiler, so any dependency without a prebuilt wheel fails to install. Either use a multi-stage build that compiles in a full image and copies the result, or accept the larger base.

Run as non-root. Containers default to root, and a compromised process with root inside the container is a bigger problem than it needs to be.

Configuration comes from environment variables at runtime, never baked into the image, so the same image runs in staging and production unchanged.

**Likely follow-ups**

- Why copy requirements before the application code?
- What happens to your logs if you buffer stdout?
- When would you use a slim base and when would you not?

---

## 53. What would you put in CI to keep a shared Python codebase consistent?

*Medium · Common*

**Short answer.** A formatter, a linter and a type checker, all run automatically. The formatter ends style arguments by making them non-negotiable, the linter catches real defects such as unused variables and shadowed names, and the type checker finds signature mismatches before runtime.

The current common stack is `ruff` for both linting and formatting, and `mypy` for types. `black` and `flake8` remain widely used and do the same jobs; ruff mostly consolidates them and runs much faster.

```
# pyproject.toml
[tool.ruff]
line-length = 88
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP"]

[tool.mypy]
python_version = "3.12"
warn_unused_ignores = true
```

Those rule codes cover pycodestyle errors, pyflakes defects, import sorting, common bug patterns from flake8-bugbear, and outdated syntax.

The formatter’s value is not aesthetic. It removes formatting from code review entirely, so nobody spends a comment on a line break, and diffs contain only real changes rather than whitespace churn from someone’s editor.

The linter catches things that are genuinely wrong: a variable assigned and never used, a name shadowing a builtin, a mutable default argument, an f-string with no placeholders. These are cheap to catch and expensive to find at runtime.

Type checking is the one people question, since annotations do nothing at runtime. What they buy you is a check that a function returning `dict | None` is not being indexed directly by its caller, which is a `TypeError` waiting for the one input that produces None. On a shared pipeline codebase where a signature change ripples through several modules, that is worth the effort.

Two practical points.

Run everything in CI and fail the build, or the standards are advisory. Run the same tools locally through `pre-commit` so failures surface before the push rather than after.

Adopting on a large existing codebase means starting permissive. Enable a small rule set, fix what it finds, and tighten. For mypy, `--ignore-missing-imports` and per-module strictness let you type the new code without a thousand errors on day one.

**Likely follow-ups**

- Type hints are optional at runtime — so what does a type checker actually buy you?
- How would you introduce this into a large existing codebase without a thousand failures?
- Should the formatter run in CI or only locally?

---

## 54. A worker process grows from 300 MB to 6 GB over a week and gets killed. How do you find the cause?

*Hard · Occasional*

**Short answer.** Something is holding references that should have been released: an unbounded cache, an ever-growing list, an accumulating logger handler. Take `tracemalloc` snapshots at intervals and compare them to see which allocation sites keep growing, rather than inspecting memory at a single point.

CPython frees an object when its reference count drops to zero, so a leak in Python almost always means something is still pointing at the data. The usual suspects are structural, not exotic.

An unbounded cache. A dict keyed on customer ID that never evicts grows with the key space forever. `functools.lru_cache` with no `maxsize` does the same thing.

An accumulating list. Appending results inside a long-running loop for a summary computed at the end, in a process that never reaches the end.

A `defaultdict` read as if it were a lookup: `if d[key]` inserts the key, so a membership test grows the dict.

Handlers or callbacks registered per iteration instead of once at startup.

To find it, compare snapshots rather than looking at one:

```
import tracemalloc

tracemalloc.start()
baseline = tracemalloc.take_snapshot()

for cycle in range(10):
    process_batch()

current = tracemalloc.take_snapshot()
for stat in current.compare_to(baseline, "lineno")[:15]:
    print(stat)
```

The output names the file and line where the growing allocations happen, sorted by size difference. That is the whole diagnosis in most cases.

A cheaper first check is counting objects by type:

```
from collections import Counter
import gc

Counter(type(o).__name__ for o in gc.get_objects()).most_common(10)
```

Run it at intervals. A type whose count climbs steadily points at what is accumulating even before you know where.

Reference cycles are the case where the counter alone does not help: two objects referring to each other never reach zero. Python’s cyclic collector handles those, so they delay rather than prevent collection, and they matter most when the objects hold external resources. `gc.collect()` returning a large number of collected objects tells you cycles are being created.

Fragmentation is worth naming too. A process that peaked at 6 GB may not return memory to the OS even after freeing it, so RSS stays high while Python’s own accounting shows the objects gone.

**Likely follow-ups**

- Python has a garbage collector — how is a leak even possible?
- What would you look at first, growth in object count or in total bytes?
- Where do reference cycles fit into this?

---

## 55. When would you define your own exception class rather than raising ValueError?

*Medium · Occasional*

**Short answer.** Define your own when callers need to distinguish your failure from everything else and handle it specifically. A shared base class for your library lets callers catch all of it in one clause. Built-ins are fine when the meaning genuinely matches, such as a bad argument value.

The test is whether a caller would want to catch it separately. If the answer is no, `ValueError` with a clear message is enough and adding a class is noise.

```
class FeedError(Exception):
    """Base for everything this module raises."""

class SchemaMismatch(FeedError):
    def __init__(self, expected, found):
        self.expected = expected
        self.found = found
        super().__init__(
            f"expected {expected} columns, found {found}"
        )

class FeedUnavailable(FeedError):
    pass
```

The base class is the part that earns its keep. A caller who does not care which specific thing went wrong writes `except FeedError:` once, and keeps working when you add a fourth subclass next month. Without a shared base, every caller maintains a growing tuple of your exception names.

Carrying structured attributes matters more than the message. A retry layer needs to know whether the failure is transient, and parsing that out of a string is fragile:

```
except SchemaMismatch as exc:
    if exc.found > exc.expected:
        drop_extra_columns()
```

Two things worth mentioning in an interview. Inherit from a built-in when the semantics genuinely match and callers may already be catching it: a custom lookup failure inheriting from `KeyError` keeps existing code working. Do not inherit from `BaseException`; that is reserved for things that should escape ordinary handlers.

And when wrapping a lower-level error, preserve the chain:

```
except ConnectionError as exc:
    raise FeedUnavailable("vendor feed down") from exc
```

`from exc` attaches the original traceback. Without it, the log shows your exception and nothing about the socket error that caused it, which is exactly the detail you need at 2 AM.

**Likely follow-ups**

- What would you put in the exception besides a message?
- Where would you inherit from something other than Exception?
- How do you keep the original error visible when you wrap it in your own?

---

## 56. pip says two of your dependencies need incompatible versions of the same library. How do you get out of it?

*Medium · Occasional*

**Short answer.** Find out which packages demand which ranges, then look for versions that overlap. Usually one package has a newer release with a widened constraint. If nothing overlaps, the options are pinning an older version of one package, splitting the workload into separate environments, or dropping a dependency.

Start by reading the actual constraints rather than guessing:

```
pip install pipdeptree
pipdeptree --warn fail
```

That shows the tree and flags conflicts with the versions each parent requires. `pip check` on an already-installed environment reports incompatibilities more briefly.

Then work through the options in order.

**Upgrade the constrainer.** Most conflicts are stale. A package pinned to an old range often has a later release that widened it, and upgrading resolves everything without further thought.

**Find the overlap.** If A wants `>=2.0,<3` and B wants `>=1.8,<2.5`, then 2.0 to 2.4 satisfies both, and pip’s resolver should find it. If it is thrashing, pin the shared dependency explicitly to give it a starting point.

**Split the environment.** Two pipeline stages needing genuinely incompatible libraries do not have to share a process. Separate virtual environments, or separate container images, remove the conflict entirely and are often cleaner than forcing a resolution.

**Drop one.** Sometimes the dependency is doing very little and replacing it with fifty lines of your own code is the right call.

What not to do is `pip install --no-deps` or force-installing a version that satisfies nobody. It appears to work, and it fails later at an import or an attribute access, at which point the cause is three weeks behind you.

Preventing recurrence is the second half of the answer. Pin exact versions in a lock file for anything deployed, so installs are reproducible and a transitive upgrade cannot arrive silently. `pip-tools` compiles a `requirements.in` of loose constraints into a fully pinned `requirements.txt`; Poetry and uv do the same with their own lock formats. Keep the loose file as the statement of intent and the lock file as the record of what actually gets installed.

**Likely follow-ups**

- What's the difference between a requirements file and a lock file here?
- When would you vendor a dependency rather than resolve the conflict?
- How do you stop this recurring on the next install?

---

More Data Engineer sets: https://codeayan.com/get-hired/data-engineer
All interview prep: https://codeayan.com/get-hired