Top 50 Python Interview Questions for Data Engineers

Detailed explanation

Choosing in data engineering code

• list: ordered batches of records, results to write in order.
• tuple: fixed-shape records such as (customer_id, order_date), composite dict keys, function returns. Immutability prevents accidental changes. NamedTuple or dataclasses add field names (see dataclasses vs Pydantic models).
• set: deduplicating IDs, fast “have we seen this?” checks, and set algebra (a - b for missing keys, a & b for overlap) when reconciling sources.
• dict: lookups (code to label), grouping (defaultdict(list)), counting (Counter), JSON-like records.

Hashability: set members and dict keys must be hashable. Lists and dicts aren’t, so convert them to tuples or frozensets. Floats like NaN behave oddly as keys (NaN != NaN).

Performance trap: if x in big_list inside a loop is O(n²). Convert the list to a set first.

Memory: tuples are slightly smaller than lists, and sets and dicts use more memory because of their hash tables. For millions of rows, use columnar structures (pandas, Polars, Arrow) instead of lists of dicts (see Processing data larger than memory in pandas (chunks, dtypes)).

Example

# Runtime: Python 3.12
from collections import Counter, defaultdict

source_ids = [101, 102, 103, 103, 104]
target_ids = {102, 104, 105}

missing_in_target = set(source_ids) - target_ids          # {101, 103}
unexpected_in_target = target_ids - set(source_ids)       # {105}

orders = [("c1", 120.0), ("c2", 40.0), ("c1", 30.0)]      # tuples as records
spend = defaultdict(float)
for customer_id, amount in orders:
    spend[customer_id] += amount                          # {'c1': 150.0, 'c2': 40.0}

status_counts = Counter(["paid", "paid", "refunded"])     # Counter({'paid': 2, 'refunded': 1})
daily_key = {("c1", "2026-03-01"): 120.0}                 # tuple as composite key
print(missing_in_target, unexpected_in_target, dict(spend), status_counts)

Real-world use case

A reconciliation script checked 2 million source IDs against a target list with in, and took 40 minutes. Converting the target IDs to a set made it finish in 2 seconds.

Common mistakes and red flags

• Membership tests on large lists.
• Using lists as dict keys.
• Assuming sets preserve order.

Follow-up questions

• Why is dict lookup O(1)? — Hash table: key hash → bucket; collisions resolved by probing.
• When is a tuple mutable in practice? — If it contains mutable objects (a tuple holding a list), the inner list can change.

In depth: List vs tuple vs set: when would you use each in a data pipeline?

Detailed explanation

Names are references. a = [1, 2] then b = a means b.append(3) changes a too. Both names point to the same list. With immutables, s = "x" then t = s then t += "y" creates a new string for t, and s stays “x”.

Function arguments are passed by object reference (“call by sharing”):

• Mutating a passed list or dict inside a function affects the caller.
• Rebinding the parameter (items = []) doesn’t affect the caller.

The default argument trap

def add_record(rec, batch=[]):     # evaluated once at def time
    batch.append(rec)
    return batch
add_record(1)  # [1]
add_record(2)  # [1, 2]  <- surprise: same list

Fix: def add_record(rec, batch=None): batch = [] if batch is None else batch.

The same issue appears with dataclass fields (use field(default_factory=list); see dataclasses vs Pydantic models), class attributes shared across instances, and module-level caches reused across Lambda or worker invocations.

Why it matters in pipelines

• Shared mutable state leaks data between batches or requests. For example, an accumulator list keeps growing across Airflow task retries in the same worker process.
• Functions that mutate their inputs cause hard-to-trace bugs. Prefer returning new objects, or document the mutation clearly.
• Immutable objects are safe to share across threads and to use as dict keys.

Example

# Runtime: Python 3.12
def normalise(records: list[dict], *, seen: set[str] | None = None) -> list[dict]:
    seen = set() if seen is None else seen
    out = []
    for r in records:
        if r["id"] in seen:
            continue
        seen.add(r["id"])
        out.append({**r, "email": r["email"].strip().lower()})   # new dict, input untouched
    return out

batch1 = [{"id": "a", "email": " [email protected] "}]
batch2 = [{"id": "a", "email": "[email protected]"}]
print(normalise(batch1), normalise(batch2))   # each call gets a fresh 'seen' set

Result: both calls return the record. With a mutable default seen=set(), the second call would wrongly drop “a”.

Real-world use case

A Lambda function used def handler(event, context, buffer=[]) to collect events, and warm invocations kept appending to the same list. Some outputs contained events from previous requests until the default was changed to None.

Common mistakes and red flags

• Mutable defaults.
• Assuming assignment copies objects.
• Mutating input dicts in shared utility functions.

Follow-up questions

• Is a tuple containing a list hashable? — No; hashing fails because the list isn’t hashable.
• How do you copy instead of share? — list(x), x.copy(), copy.deepcopy(x) (see Shallow copy vs deep copy).

In depth: Why are mutable default arguments dangerous in Python?

Detailed explanation

Illustration

original = {"id": 1, "tags": ["a", "b"]}
shallow  = original.copy()     -> new dict, SAME "tags" list
deep     = copy.deepcopy(original) -> new dict, NEW "tags" list
shallow["tags"].append("c")    -> original["tags"] is now ["a","b","c"]
deep["tags"].append("d")       -> original unaffected

Ways to copy

Performance: deepcopy is slow for big structures, because it visits every object. In data pipelines, it’s usually better to:

• Build new structures with comprehensions ({**r, "x": new} creates a new top-level dict, sharing unchanged nested values deliberately).
• Use immutable records (frozen dataclasses, tuples), so copying is unnecessary.
• Use DataFrames, whose copy() semantics are explicit. pandas 2.x copy-on-write mode avoids many accidental-mutation issues, and pandas 3.0 makes copy-on-write the default (see pandas: loc vs iloc, and vectorisation vs apply).

Common places this bites

• Reusing a base config dict for several jobs and mutating nested sections.
• Building “template” records and appending to their nested lists.
• [[0] * 3] * 3 creates three references to the same inner list. Use [[0] * 3 for _ in range(3)].

Example

# Runtime: Python 3.12
import copy

base_config = {"source": "s3://lake/raw/", "options": {"format": "parquet", "columns": ["id"]}}

job_a = copy.deepcopy(base_config)
job_a["options"]["columns"].append("amount")

job_b = base_config.copy()                 # shallow
job_b["options"]["columns"].append("ts")   # mutates base_config too!

print(base_config["options"]["columns"])   # ['id', 'ts']
print(job_a["options"]["columns"])         # ['id', 'amount']

grid_bad = [[0] * 2] * 2; grid_bad[0][0] = 9           # [[9, 0], [9, 0]]
grid_ok = [[0] * 2 for _ in range(2)]; grid_ok[0][0] = 9  # [[9, 0], [0, 0]]

Real-world use case

An ingestion framework generated per-table configs from a shared template with .copy(). Adding a column filter for one table silently added it to all of them. Switching to deepcopy, and later to frozen dataclass configs, fixed it.

Common mistakes and red flags

• Believing .copy() is deep.
• Using deepcopy on huge data in hot loops.
• The [[x]*n]*m trap.

Follow-up questions

• How does deepcopy handle cycles? — A memo dict tracks already-copied objects.
• How do you customise copying? — Implement __copy__/__deepcopy__.

In depth: Explain shallow copy vs deep copy.

Detailed explanation

Forms

Why they’re faster: the loop runs in optimised bytecode with a specialised append, and there’s no repeated attribute lookup of .append. Python 3.12 inlined comprehensions (PEP 709), which speeds them up further.

Readability rules of thumb

• One for and one if is fine. Two fors is the limit. Beyond that, write a loop.
• Don’t use comprehensions for side effects ([print(x) for x in xs]). Use a loop.
• Use the walrus operator to avoid computing twice: [y for x in xs if (y := parse(x)) is not None].
• Use generator expressions for aggregates over large data: sum(r["amount"] for r in records) avoids building a list.

Scope: comprehension variables don’t leak into the enclosing scope in Python 3.

Example

# Runtime: Python 3.12
from datetime import date

rows = [
    {"id": "1", "amount": "10.50", "status": "paid", "dt": "2026-03-01"},
    {"id": "2", "amount": "",      "status": "void", "dt": "2026-03-01"},
    {"id": "3", "amount": "7.25",  "status": "paid", "dt": "2026-03-02"},
]

paid_amounts = [float(r["amount"]) for r in rows if r["status"] == "paid" and r["amount"]]
by_id = {r["id"]: r for r in rows}
statuses = {r["status"] for r in rows}
dates = sorted({date.fromisoformat(r["dt"]) for r in rows})
total = sum(float(r["amount"]) for r in rows if r["amount"])      # generator, no list

print(paid_amounts, list(by_id), statuses, dates, total)
# [10.5, 7.25] ['1', '2', '3'] {'paid', 'void'} [datetime.date(2026, 3, 1), datetime.date(2026, 3, 2)] 17.75

Real-world use case

A code review replaced a 25-line nested loop that built a lookup dict, with flag variables, by a two-line dict comprehension plus a helper function. It was easier to read, easier to test, and about 30% faster.

Common mistakes and red flags

• Triple-nested comprehensions.
• Comprehensions for side effects.
• Building big lists just to sum them.

Follow-up questions

• Comprehension vs map/filter? — Comprehensions are usually clearer; map with built-ins can be marginally faster (see lambda, map, filter and reduce).
• How do you flatten a list of lists? — [x for sub in nested for x in sub] or itertools.chain.from_iterable.

Detailed explanation

Parameter order: def f(pos_only, /, normal, *args, kw_only, **kwargs).

Good uses

• Decorators and wrappers that forward any arguments (see Decorators (with arguments and functools.wraps)).
• Thin adapters around library calls, passing configuration dicts: client.get(url, **request_options).
• Keyword-only flags for clarity: def load(path, *, overwrite=False). Callers must write overwrite=True, which prevents positional mistakes.

Overuse warnings

• def run(**kwargs) everywhere hides the real interface. IDEs and type checkers can’t help, and typos in keys pass silently. Prefer explicit parameters, or a typed config object (dataclass or Pydantic; see dataclasses vs Pydantic models).
• Typing: *args: int means each positional is int, and **kwargs: str means each value is str. For precise kwargs typing, use Unpack[TypedDict] (PEP 692).

Rule of thumb: use *args/**kwargs when you’re genuinely forwarding arguments you don’t need to understand (wrappers, adapters). When you’re the final consumer, spell the parameters out.

Example

# Runtime: Python 3.12
from typing import Any

def build_query(table: str, /, *columns: str, where: str | None = None, limit: int | None = None) -> str:
    cols = ", ".join(columns) or "*"
    sql = f"SELECT {cols} FROM {table}"
    if where:
        sql += f" WHERE {where}"
    if limit is not None:
        sql += f" LIMIT {int(limit)}"
    return sql

def log_call(func, *args: Any, **kwargs: Any):
    print(f"calling {func.__name__} args={args} kwargs={kwargs}")
    return func(*args, **kwargs)

defaults = {"limit": 100}
overrides = {"where": "status = 'paid'"}
print(log_call(build_query, "orders", "order_id", "amount", **{**defaults, **overrides}))
# SELECT order_id, amount FROM orders WHERE status = 'paid' LIMIT 100

This builder is for trusted, code-defined identifiers and conditions only. Values that come from users must go through parameterised queries (see Querying databases from Python safely (parameterised queries, SQLAlchemy)).

Real-world use case

A loader function had the signature load(df, table, True, False, 500), and positional booleans caused a production overwrite when the arguments were swapped. Changing it to load(df, table, *, overwrite=False, create=False, batch_size=500) made calls self-documenting.

Common mistakes and red flags

• **kwargs hiding typos.
• Positional boolean flags.
• Forgetting that dict-merge order determines which value wins.

Follow-up questions

• What type is args? — tuple; kwargs is dict.
• Why positional-only params? — Freedom to rename them later without breaking callers; matches C-implemented built-ins.

Detailed explanation

Interfaces in Python

• ABCs (abc.ABC, @abstractmethod): enforce that subclasses implement methods, failing at instantiation time.
• typing.Protocol: structural typing (“anything with read() and close()”). No inheritance needed, and type checkers verify it (see Type hints and static checking).
• Duck typing: just call the method. Simple, but errors surface at runtime.

Design guidance for DE code

• Keep hierarchies shallow (one level of abstract base plus concrete classes).
• Inject dependencies (clients, configs) through constructors rather than creating them inside, which helps testing.
• Avoid “god” base classes with dozens of hooks.
• Multiple inheritance and mixins: use them sparingly, and understand the MRO (method resolution order).
• Often a few plain functions plus dataclasses beat classes entirely. Don’t use OOP for its own sake.

Example

# Runtime: Python 3.12
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Iterable, Iterator, Protocol

class Source(ABC):                                  # inheritance for a shared interface
    @abstractmethod
    def read(self) -> Iterator[dict]: ...

class CsvSource(Source):
    def __init__(self, rows: list[dict]):
        self._rows = rows
    def read(self) -> Iterator[dict]:
        yield from self._rows

class Sink(Protocol):                               # structural interface
    def write(self, records: Iterable[dict]) -> int: ...

class MemorySink:
    def __init__(self):
        self.records: list[dict] = []
    def write(self, records: Iterable[dict]) -> int:
        batch = list(records); self.records.extend(batch); return len(batch)

@dataclass
class Pipeline:                                     # composition: has a source and a sink
    source: Source
    sink: Sink
    def run(self) -> int:
        cleaned = ({**r, "email": r["email"].lower()} for r in self.source.read())
        return self.sink.write(cleaned)

sink = MemorySink()
print(Pipeline(CsvSource([{"email": "[email protected]"}]), sink).run(), sink.records)  # 1 [{'email': '[email protected]'}]

Real-world use case

A connector framework had a 5-level inheritance tree, and changing a base method broke 20 connectors. Refactoring to composition (a Pipeline with injected Source, Transform and Sink objects behind Protocols) let teams add connectors without touching shared base classes, and unit tests used in-memory sinks.

Common mistakes and red flags

• Deep inheritance for code reuse.
• Creating clients inside classes, which makes them untestable.
• Classes where functions would do.

Follow-up questions

• What’s MRO? — The order Python searches classes for attributes (C3 linearization); Class.__mro__.
• ABC vs Protocol? — ABC requires explicit inheritance and runtime enforcement; Protocol is structural and checked statically.

Detailed explanation

Why classmethod factories

• They return cls(...), so subclasses get instances of themselves. SnowflakeConfig.from_env() returns a SnowflakeConfig, not the base class.
• They keep construction logic (parsing env vars, files, dicts) out of __init__, so __init__ stays simple and testable with explicit values.

Static methods: use sparingly

• Fine for small helpers tied to the class’s domain (Money.round_half_even(x)).
• Often a module-level function is clearer and easier to import and test. Python isn’t Java, and not everything needs a class.

Related: @property gives computed attributes accessed without parentheses (order.net_amount), and functools.cached_property computes once per instance.

Interview angle: expect a follow-up asking which method type fits a scenario. For example, “parse a config from a YAML file” is a classmethod factory, “validate an identifier format” is a static helper or module function, and “compute this record’s net amount” is an instance method or property. Explaining why, especially how subclassing works with cls, shows real understanding rather than memorised definitions.

Example

# Runtime: Python 3.12
import os
from dataclasses import dataclass

@dataclass(frozen=True)
class WarehouseConfig:
    account: str
    warehouse: str
    database: str
    role: str = "TRANSFORMER"

    @classmethod
    def from_env(cls, prefix: str = "SF_") -> "WarehouseConfig":
        return cls(
            account=os.environ[f"{prefix}ACCOUNT"],
            warehouse=os.environ.get(f"{prefix}WAREHOUSE", "TRANSFORM_XS"),
            database=os.environ[f"{prefix}DATABASE"],
        )

    @staticmethod
    def normalise_identifier(name: str) -> str:
        return name.strip().upper()

    def fully_qualified(self, schema: str, table: str) -> str:          # instance method
        n = self.normalise_identifier
        return f"{n(self.database)}.{n(schema)}.{n(table)}"

os.environ.update({"SF_ACCOUNT": "acme-xy123", "SF_DATABASE": "analytics"})
cfg = WarehouseConfig.from_env()
print(cfg.fully_qualified("marts", "fct_orders"))      # ANALYTICS.MARTS.FCT_ORDERS

Real-world use case

A config class’s __init__ read environment variables directly, so tests had to set dozens of env vars. Moving that logic to from_env() let tests construct configs explicitly, while production code kept calling from_env().

Common mistakes and red flags

• Static methods that need class info.
• __init__ doing I/O.
• Classes full of static methods that should be a module.

Follow-up questions

• Why not just use a module function as factory? — Fine too; classmethods integrate with subclassing.
• What does @property add? — Attribute-style access with computed/validated values.

Detailed explanation

Structure

try:            # only the risky call
    ...
except SpecificError as e:   # handle what you understand
    ...
else:           # runs if no exception (keep happy path out of try)
    ...
finally:        # always runs (cleanup)
    ...

Principles

1. Specific over broad. except (TimeoutError, ConnectionError) rather than except Exception. A bare except: even catches KeyboardInterrupt and SystemExit.
2. Fail loudly by default. A pipeline that “succeeds” after silently skipping errors produces wrong data, which is worse than failing.
3. Add context and chain. raise LoadError(f"failed loading {table}") from e keeps the original traceback (see Custom exceptions and retryable errors).
4. Log once, at the right level. Don’t log and re-raise at every layer, which duplicates noise. Use log.exception() inside except to include the traceback.
5. Classify errors. Transient errors (timeouts, 429, 5xx) can be retried (see Retries with exponential backoff and rate limiting). Data errors (bad record): quarantine and continue, if the business allows (see Coding: validate records and split good from bad). Programming errors (KeyError from a bug) must crash.
6. Clean up with with or finally: connections, temp files, locks (see Context managers and the with statement).
7. Exception groups (3.11+, except*) handle several concurrent failures, for example from asyncio.TaskGroup (see Concurrent API calls with asyncio).

Anti-patterns

Example

# Runtime: Python 3.12
import json
import logging

log = logging.getLogger(__name__)

class SourceDataError(Exception):
    """Raised when an input file cannot be interpreted."""

def load_manifest(path: str) -> dict:
    try:
        with open(path, encoding="utf-8") as f:
            raw = f.read()
    except FileNotFoundError as e:
        raise SourceDataError(f"manifest missing: {path}") from e

    try:
        manifest = json.loads(raw)
    except json.JSONDecodeError as e:
        raise SourceDataError(f"manifest is not valid JSON: {path} (line {e.lineno})") from e
    else:
        if "files" not in manifest:
            raise SourceDataError(f"manifest has no 'files' key: {path}")
        return manifest

try:
    load_manifest("landing/2026-03-01/_manifest.json")
except SourceDataError:
    log.exception("cannot start load")      # logs traceback with chained cause
    raise SystemExit(2)

Real-world use case

An ingestion job wrapped each table load in try/except Exception: continue. For two weeks, 3 of 40 tables failed silently because a credential had expired, while the job reported success. Replacing that with specific handling plus a final failure status when any table failed restored visibility.

Common mistakes and red flags

• Bare except.
• Swallowed errors.
• Losing the original traceback.

Follow-up questions

• EAFP vs LBYL? — “Easier to ask forgiveness” (try/except) vs “look before you leap” (checks); EAFP is idiomatic for races like file existence.
• What does raise alone do in except? — Re-raises the current exception with original traceback.

In depth: How should exceptions be handled in production data pipelines?

Detailed explanation

CSV with the standard library

• open(path, newline="", encoding="utf-8"). newline="" prevents blank lines on Windows and handles embedded newlines correctly.
• csv.DictReader gives dicts keyed by header. Everything is a string, so convert explicitly.
• Dialects: delimiter, quotechar, quoting=csv.QUOTE_MINIMAL.
• BOM from Excel exports: read with encoding="utf-8-sig".

JSON and JSONL

• json.load(f) for whole documents, which loads everything into memory.
• JSON Lines (one object per line) streams well: for line in f: json.loads(line).
• json.dumps(obj, default=str) handles datetimes and decimals by converting them to strings. Better to serialise deliberately (isoformat, str for Decimal).
• Large single JSON arrays: use streaming parsers (ijson), or convert upstream to JSONL.

Parquet (pyarrow, pandas, Polars)

• pd.read_parquet(path, columns=[...], filters=[...]) and df.to_parquet(path, index=False, compression="zstd").
• pl.read_parquet/scan_parquet, df.write_parquet.
• pyarrow.dataset for partitioned directories (dataset("s3://...", partitioning="hive")).
• Preserves types (timestamps, decimals, nested), compresses well, and supports column pruning (see Row-oriented vs columnar file formats).

Encoding and correctness checklist

• Always set encoding explicitly (UTF-8).
• Make timezone handling explicit for timestamps (see datetime and time zones with zoneinfo).
• Money: Decimal or integer cents, never float for exact amounts in finance outputs.
• Atomic writes: write to a temp file or prefix, then rename or move (see Structure of an idempotent, restartable ETL script).

Example

# Runtime: Python 3.12 with pandas 2.2, pyarrow 17+
import csv
import json
from decimal import Decimal
from pathlib import Path
import pandas as pd

src = Path("orders.csv")
src.write_text("order_id,amount,currency\n1,19.99,USD\n2,5.00,EUR\n", encoding="utf-8")

with src.open(newline="", encoding="utf-8-sig") as f:
    rows = [{"order_id": int(r["order_id"]), "amount": Decimal(r["amount"]), "currency": r["currency"]}
            for r in csv.DictReader(f)]

with Path("orders.jsonl").open("w", encoding="utf-8") as f:
    for r in rows:
        f.write(json.dumps({**r, "amount": str(r["amount"])}) + "\n")

df = pd.read_json("orders.jsonl", lines=True, dtype={"amount": "string"})
df["amount"] = df["amount"].map(Decimal)
pd.DataFrame(rows).astype({"amount": "float64"}).to_parquet("orders.parquet", index=False, compression="zstd")
print(pd.read_parquet("orders.parquet", columns=["order_id", "amount"]))

Converting to float64 for Parquet is fine for analytics. For exact money, store integer cents or a pyarrow decimal type instead.

Real-world use case

A partner’s CSVs were exported from Excel with a BOM, so the first column became \ufefforder_id and every lookup failed. Reading with utf-8-sig, plus a header validation step, fixed it.

Common mistakes and red flags

• No encoding specified.
• Floats for money in exports.
• json.load on multi-GB files.

Follow-up questions

• Why JSONL over JSON arrays? — Streamable, appendable, splittable by line.
• How do you write a partitioned Parquet dataset? — pyarrow.dataset.write_dataset(..., partitioning=["event_date"]) or Polars/pandas partition_cols.

Detailed explanation

Why environment variables: the same code runs in dev, CI and prod, with only the environment differing (the twelve-factor app principle). They work across containers, Lambda, Airflow, Kubernetes and Databricks jobs.

Patterns

Security practices

• Never commit secrets. Add .env to .gitignore, and use pre-commit secret scanners (gitleaks, detect-secrets).
• Prefer short-lived credentials and identity over static keys: IAM roles, workload identity, OIDC from CI, key-pair auth with rotation (see How do you manage secrets in pipelines?).
• Don’t print env vars or full configs in logs. Mask secret fields (SecretStr in Pydantic shows **********).
• Restrict who can read secrets in the orchestrator and cloud.
• Environment variables can leak through process listings or crash dumps on some systems. For highly sensitive secrets, fetch them from the manager at runtime into memory.

Validate at startup: fail immediately with a clear message if a required setting is missing or malformed, rather than crashing halfway through a load.

Example

# Runtime: Python 3.12 with pydantic-settings 2.x
from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class EtlSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="ETL_", env_file=".env", extra="ignore")

    env: str = Field(pattern="^(dev|ci|prod)$")
    source_bucket: str
    batch_size: int = Field(default=5000, ge=100, le=100_000)
    snowflake_account: str
    snowflake_user: str
    snowflake_private_key: SecretStr          # injected by secrets manager in prod

settings = EtlSettings()                      # raises ValidationError listing all problems
print(settings.env, settings.batch_size, settings.snowflake_private_key)   # key prints as **********
key_pem = settings.snowflake_private_key.get_secret_value()                # use only where needed

export ETL_ENV=dev ETL_SOURCE_BUCKET=acme-dev-landing \
       ETL_SNOWFLAKE_ACCOUNT=acme-xy123 ETL_SNOWFLAKE_USER=SVC_ETL_DEV
export ETL_SNOWFLAKE_PRIVATE_KEY="$(aws secretsmanager get-secret-value --secret-id etl/dev/sf-key --query SecretString --output text)"

Real-world use case

A database password committed in a config file was found by a public repository scanner within hours of a repo being made public by mistake. The team rotated it, moved all secrets to the cloud secrets manager injected at runtime, and added gitleaks to pre-commit and CI.

Common mistakes and red flags

• Hard-coded credentials.
• Committing .env files.
• Logging the config dict.
• Converting env strings without validation ("false" is truthy).

Follow-up questions

• Why is bool(os.getenv("FLAG")) wrong? — Any non-empty string, including “false”, is True; parse explicitly.
• How do Airflow tasks get secrets? — Connections/Variables backed by a secrets backend (see How do you manage secrets in pipelines?).

Detailed explanation

Why is “works” sometimes

• CPython caches small ints (-5 to 256), so a = 256; b = 256; a is b is True, but with 257 it may be False.
• Some strings are interned (identifier-like literals), so is can be True for literals and False for computed strings.
• These are implementation details. Relying on them gives bugs that appear only with certain data. Python warns: SyntaxWarning: "is" with a literal.

None checks

• x is None is correct and fast, and immune to custom __eq__.
• x == None can be overridden. pandas and NumPy return element-wise arrays, for example, and linters flag it.
• In pandas, use isna() / notna(). NaN is never equal to itself (NaN == NaN is False), and None in object columns behaves differently from NaN in float columns.

Truthiness vs None: if not x: treats 0, "", [] and None the same. Use if x is None: when 0 or empty are valid values, such as an amount of 0 or an empty string from the source.

Sentinels: _MISSING = object() distinguishes “not provided” from None in function arguments: def get(key, default=_MISSING), then if default is _MISSING:.

Example

# Runtime: Python 3.12
import math

a = int("1000"); b = int("1000")
print(a == b, a is b)                 # True False  (value equal, different objects)

def describe(discount: float | None) -> str:
    if discount is None:              # 0.0 is a valid discount, so don't use 'if not discount'
        return "no discount info"
    return f"discount={discount}"

print(describe(0.0), describe(None))  # discount=0.0  no discount info

nan = float("nan")
print(nan == nan, math.isnan(nan))    # False True

_MISSING = object()
def get_setting(cfg: dict, key: str, default=_MISSING):
    if key in cfg:
        return cfg[key]
    if default is _MISSING:
        raise KeyError(key)
    return default
print(get_setting({"a": None}, "a"), get_setting({}, "b", default=None))   # None None

Real-world use case

A filter used if not row["discount"]: row["discount"] = DEFAULT, which overwrote legitimate 0% discounts with the default 5%, and reported revenue was overstated. Changing it to is None fixed it.

Common mistakes and red flags

• x is 5 or name is "abc".
• == None.
• Truthiness checks where 0 or empty are valid.

Follow-up questions

• What does id() return? — An integer identity (memory address in CPython) unique for the object’s lifetime.
• Why NaN != NaN? — IEEE 754 defines NaN as unequal to everything, including itself.

Detailed explanation

Steps

1. Normalise: text.lower(), and decide on unicode handling (casefold() for aggressive case-insensitivity).
2. Tokenise: re.findall(r"[a-z0-9']+", line), or \w+ for unicode words. Handle apostrophes deliberately.
3. Filter: stopwords, minimum length, numbers.
4. Count: Counter.update(tokens) per line, so memory is bounded by the vocabulary, not the text size.
5. Top-k: most_common(k) is O(n log k) with heapq.nlargest. Full sorting is O(n log n).
6. Ties: decide on deterministic output (count descending, then word ascending), and implement it explicitly with heapq.nsmallest(k, items, key=lambda kv: (-kv[1], kv[0])).

Complexity: O(N) to tokenise and count N tokens, O(V log k) for top-k over V unique words, and O(V) memory.

Scaling discussion

• Bigger than one machine: map (emit word, 1) → shuffle by word → reduce (sum). That’s Spark groupBy("word").count() or reduceByKey (see groupByKey vs reduceByKey (RDD API), Coding: word count / frequency count).
• Approximate top-k with limited memory: Count-Min Sketch plus a heap, or Space-Saving / Misra–Gries algorithms for streams.
• Parallel on one machine: split files across processes, count per process, then merge Counters (c1 + c2 or update).

Example

# Runtime: Python 3.12
import heapq
import re
from collections import Counter
from typing import Iterable

TOKEN = re.compile(r"[a-z0-9']+")
STOPWORDS = {"the", "a", "an", "and", "of", "to", "is"}

def word_counts(lines: Iterable[str]) -> Counter[str]:
    counts: Counter[str] = Counter()
    for line in lines:
        counts.update(w for w in TOKEN.findall(line.lower()) if w not in STOPWORDS and len(w) > 1)
    return counts

def top_k(counts: Counter[str], k: int) -> list[tuple[str, int]]:
    # deterministic: highest count first, then alphabetical
    return heapq.nsmallest(k, counts.items(), key=lambda kv: (-kv[1], kv[0]))

text = [
    "Timeout error while calling the payments API.",
    "Retry succeeded after timeout; payments API slow.",
    "Error: invalid token. Error persisted.",
]
c = word_counts(text)
print(top_k(c, 3))      # [('error', 3), ('api', 2), ('payments', 2)]

payments and timeout both appear twice. Alphabetical tie-breaking puts payments third, before timeout.

Real-world use case

An on-call script scanned the last hour of error logs line by line and printed the top 10 tokens. That surfaced “timeout” and a specific endpoint name within seconds of a partner outage.

Common mistakes and red flags

• text.split() leaving punctuation attached.
• Sorting the whole vocabulary for small k.
• Non-deterministic tie order.

Follow-up questions

• Complexity of most_common(k)? — O(n log k) using a heap when k is given.
• How do you merge results from multiple workers? — Sum Counters (associative, commutative).

Detailed explanation

Two-sum

• Brute force checks all pairs, O(n²).
• Hash map: for i, x in the list, if target - x is in seen, return (seen[target - x], i). Otherwise seen[x] = i. That’s O(n) time and O(n) space.
• Variants: return all pairs (handle duplicates), sorted input (two pointers, O(1) space), or counting pairs (a frequency map).
• Clarify the edge cases: no solution, multiple solutions, using the same element twice (not allowed), negative numbers.

Group anagrams

• Key options:
  • "".join(sorted(word)): O(m log m) per word of length m.
  • A count tuple over a fixed alphabet: O(m) per word. Works for lowercase a–z only.
• defaultdict(list) groups the words. The result is list(groups.values()).
• Clarify: case sensitivity, spaces and punctuation, unicode.

Why interviewers ask these in DE interviews: they test whether you reach for hashing to turn quadratic scans into linear passes. That’s the same instinct needed for joins, dedupe, reconciliation and hash-partitioning in distributed systems.

Data engineering parallels

• Two-sum is like finding matching debit and credit entries that net to zero (reconciliation), or pairing request and response events.
• Group anagrams is like grouping records by a normalised key (canonical names, fuzzy-match blocking keys) before deduplication.

Example

# Runtime: Python 3.12
from collections import defaultdict

def two_sum(nums: list[int], target: int) -> tuple[int, int] | None:
    seen: dict[int, int] = {}
    for i, x in enumerate(nums):
        j = seen.get(target - x)
        if j is not None:
            return j, i
        seen[x] = i
    return None

def group_anagrams(words: list[str]) -> list[list[str]]:
    groups: dict[str, list[str]] = defaultdict(list)
    for w in words:
        key = "".join(sorted(w.lower().replace(" ", "")))
        groups[key].append(w)
    return list(groups.values())

print(two_sum([3, 8, 11, 4, -2], 9))                    # (2, 4) -> 11 + (-2)
print(two_sum([1, 2, 3], 7))                             # None
print(group_anagrams(["listen", "silent", "enlist", "google", "Elgoog", "cat"]))
# [['listen', 'silent', 'enlist'], ['google', 'Elgoog'], ['cat']]

# Reconciliation flavour: find ledger entries that net to zero
ledger = [("t1", 250), ("t2", -100), ("t3", -250), ("t4", 40)]
by_amount: dict[int, str] = {}
for txn, amt in ledger:
    if -amt in by_amount:
        print("offsetting pair:", by_amount[-amt], txn)  # offsetting pair: t1 t3
    by_amount[amt] = txn

Real-world use case

A finance reconciliation script compared 400,000 debits against 400,000 credits with nested loops, and ran overnight. A dict keyed by (account, amount) matched them in under 2 seconds.

Common mistakes and red flags

• Nested loops.
• Using the same element twice.
• Not clarifying normalisation for anagrams.

Follow-up questions

• What if input is sorted? — Two pointers from both ends, O(n) time and O(1) space.
• What’s the complexity of the sorted-key anagram approach? — O(n · m log m).

Detailed explanation

Approaches

Choosing batch size

• Respect API limits (for example 100 records per request) and payload size limits.
• Database bulk loads: thousands to tens of thousands of rows per executemany or COPY, or better, write files and use COPY (see Querying databases from Python safely (parameterised queries, SQLAlchemy)).
• Memory: batch size × record size must fit comfortably.
• Latency vs throughput: bigger batches mean fewer calls but bigger retries when a batch fails.

Failure handling: if a batch fails, decide whether to retry the whole batch (idempotent sink), split it to isolate bad records (binary search on the batch), or quarantine it (see Coding: validate records and split good from bad). Track batch numbers for checkpointing and resume (see Structure of an idempotent, restartable ETL script).

Example

# Runtime: Python 3.12
from itertools import batched, islice
from typing import Iterable, Iterator, TypeVar

T = TypeVar("T")

def chunked(iterable: Iterable[T], n: int) -> Iterator[list[T]]:
    """Portable lazy batching (pre-3.12)."""
    if n < 1:
        raise ValueError("n must be >= 1")
    it = iter(iterable)
    while batch := list(islice(it, n)):
        yield batch

ids = range(1, 11)
print(list(batched(ids, 4)))     # [(1, 2, 3, 4), (5, 6, 7, 8), (9, 10)]
print(list(chunked(ids, 4)))     # [[1, 2, 3, 4], [5, 6, 7, 8], [9, 10]]

def send_batches(records: Iterable[dict], batch_size: int = 100) -> int:
    sent = 0
    for batch_no, batch in enumerate(batched(records, batch_size), start=1):
        payload = {"batch": batch_no, "records": list(batch)}
        # client.post("/bulk", json=payload)  # API limit: 100 records per call
        sent += len(payload["records"])
    return sent

print(send_batches(({"id": i} for i in range(250))))    # 250 (3 calls: 100, 100, 50)

Real-world use case

A CRM sync sent 1 record per API call, 2 million calls in total, and kept hitting rate limits. Batching 200 per call (the API maximum) with itertools.batched cut it to 10,000 calls and finished within the nightly window.

Common mistakes and red flags

• Slicing generators (not possible).
• Forgetting the last partial batch.
• Loading everything into a list just to slice it.

Follow-up questions

• What does batched yield? — Tuples (not lists).
• How do you parallelise batches? — Submit batches to a thread pool with bounded in-flight futures (see concurrent.futures executors).

Detailed explanation

lambda

• Expression only: no statements, annotations or docstrings.
• Good uses: sorted(rows, key=lambda r: (r["country"], -r["revenue"])), max(items, key=lambda x: x.ts), defaultdict(lambda: 0).
• Avoid assigning lambdas to names (f = lambda x: ...). Use def, which gives a better name in tracebacks (PEP 8 recommends this).
• Late binding in loops: [lambda: i for i in range(3)] all return 2. Bind with a default argument (lambda i=i: i) or use functools.partial.

map / filter

• Lazy iterators in Python 3. Wrap them in list() to materialise.
• map(str.strip, lines) is clean with existing functions. [s.strip() for s in lines] is equally clear.
• filter(None, items) drops falsy values. Careful: it also drops 0 and "".

reduce

• reduce(lambda acc, x: acc + x, nums, 0) is just sum(nums).
• Legitimate uses: combining many DataFrames (reduce(DataFrame.unionByName, dfs) in PySpark), composing functions, merging dicts or Counters, applying a sequence of transformations.
• Always provide an initial value, so empty inputs don’t raise TypeError.

Helpers: operator.itemgetter("key"), attrgetter("ts") and methodcaller("lower") are often faster and clearer than lambdas. functools.partial(f, x=1) pre-binds arguments.

Connection to distributed processing: map and reduce are the conceptual roots of MapReduce and Spark (map → shuffle → reduce; see groupByKey vs reduceByKey (RDD API)). Associative reduce functions let partial results combine in any order across partitions.

Example

# Runtime: Python 3.12
from collections import Counter
from functools import partial, reduce
from operator import itemgetter

rows = [
    {"country": "IN", "revenue": 120.0, "status": "paid"},
    {"country": "US", "revenue": 300.0, "status": "refunded"},
    {"country": "IN", "revenue": 80.0,  "status": "paid"},
]

ranked = sorted(rows, key=lambda r: (r["country"], -r["revenue"]))
paid_revenue = list(map(itemgetter("revenue"), filter(lambda r: r["status"] == "paid", rows)))
same_with_comprehension = [r["revenue"] for r in rows if r["status"] == "paid"]

monthly_counts = [Counter({"IN": 2}), Counter({"US": 1}), Counter({"IN": 1, "UK": 3})]
total_counts = reduce(lambda a, b: a + b, monthly_counts, Counter())

def apply_tax(amount: float, rate: float) -> float:
    return round(amount * (1 + rate), 2)
with_gst = list(map(partial(apply_tax, rate=0.18), paid_revenue))

funcs = [lambda i=i: i * 10 for i in range(3)]          # default arg avoids late binding
print(ranked[0], paid_revenue == same_with_comprehension,
      total_counts == Counter({"IN": 3, "US": 1, "UK": 3}), with_gst, [f() for f in funcs])
# {'country': 'IN', 'revenue': 120.0, 'status': 'paid'} True True [141.6, 94.4] [0, 10, 20]

Real-world use case

A PySpark job unioned 365 daily DataFrames with a Python loop of df = df.union(x). Using functools.reduce(DataFrame.unionByName, dfs), with allowMissingColumns=True through partial, made the intent clear and handled schema drift between days (see Driver OOM vs executor OOM: causes and fixes for the plan-size caution).

Common mistakes and red flags

• Lambda late-binding bugs.
• reduce without an initial value.
• filter(None, ...) dropping valid zeros.

Follow-up questions

• Are map/filter faster than comprehensions? — Sometimes marginally with built-in functions; with lambdas, comprehensions are usually as fast or faster.
• Where did reduce move in Python 3? — To functools.

Detailed explanation

Protocol

• iter(iterable) returns an iterator. next(iterator) returns the next item, or raises StopIteration.
• A for loop calls iter() once, then next() repeatedly.
• Iterators are themselves iterable: iter(it) is it.

Iterable vs iterator

Generators

• A function with yield returns a generator object when called. Its code runs only when you iterate it, pausing at each yield and keeping its local state.
• Generator expressions: (line.strip() for line in f).
• Benefits: constant memory for large or infinite streams, composable pipelines (see Building generator pipelines with yield from), and lazy evaluation (stop early without computing the rest).
• return inside a generator ends iteration. The value is attached to StopIteration.value.
• Extras: send(), throw() and close() for coroutine-style use. Rarely needed in DE code.

Data engineering uses

• Streaming large files line by line or in chunks (see Read a huge file in chunks with a generator).
• Paginating APIs: yield one page or record at a time (see API ingestion with pagination (offset, cursor, next-link)).
• Batching (see Coding: batch an iterable into chunks), transforms and filters chained without loading everything into memory.

Gotchas

• Exhaustion: iterating a generator twice gives nothing the second time. Materialise it with list() if you need several passes, and accept the memory cost.
• Lazy errors: exceptions surface when items are consumed, not when the generator is created.
• Open resources: a generator reading a file should use with inside, so the file closes when the generator finishes or is closed.

Example

# Runtime: Python 3.12
from typing import Iterator

def read_valid_lines(path: str) -> Iterator[str]:
    with open(path, encoding="utf-8") as f:
        for line in f:                      # file object is an iterator
            line = line.strip()
            if line and not line.startswith("#"):
                yield line

lines = read_valid_lines("events.log")      # nothing read yet
first_10 = [next(lines) for _ in range(10)] # reads only as much as needed

nums = iter([1, 2, 3])
print(list(nums), list(nums))               # [1, 2, 3] []  <- exhausted

If the file has fewer than 10 valid lines, next() raises StopIteration inside the comprehension. Use itertools.islice(lines, 10) to take up to 10 safely.

Real-world use case

A script loaded a 30 GB log file with f.readlines() and was killed by the OOM killer. Rewriting it as a generator pipeline (read, filter, parse, batch-write) ran in under 200 MB of memory.

Common mistakes and red flags

• Iterating a generator twice.
• Calling list() on huge generators “to debug”.
• Confusing generators with lists.

Follow-up questions

• How do you take the first N items lazily? — itertools.islice(gen, N).
• Can a class be an iterator? — Yes: implement __iter__ returning self and __next__.

Detailed explanation

Mechanics

• @timed above def f(): ... is the same as f = timed(f).
• The wrapper usually accepts *args, **kwargs (see *args and **kwargs), calls the original, and returns its result.
• Closures: the wrapper “remembers” variables from the enclosing scope (the decorator’s arguments, the wrapped function).

With arguments (three levels)

def retry(times):            # factory, receives decorator args
    def decorator(func):     # actual decorator
        @wraps(func)
        def wrapper(*a, **kw):   # replaces func
            ...
        return wrapper
    return decorator

functools.wraps: copies __name__, __doc__, __module__ and __qualname__, and sets __wrapped__. Without it, logs, stack traces, Airflow task names and documentation tools show “wrapper” everywhere, and inspect.signature misreports the signature.

Useful built-in decorators: @functools.lru_cache / @cache (memoisation; see Coding: implement an LRU cache), @property, @classmethod, @staticmethod (see Instance vs @classmethod vs @staticmethod), @dataclass (see dataclasses vs Pydantic models), @contextlib.contextmanager (see Context managers and the with statement).

DE uses: retrying flaky API or DB calls (see Retries with exponential backoff and rate limiting), timing and logging task steps (see Logging in pipelines), validating inputs, registering transforms in a plugin registry, and Airflow’s TaskFlow @task / @dag.

Cautions

• Stacking order matters. @a @b def f means a(b(f)).
• Decorators on methods receive self in args.
• Async functions need async wrappers (async def wrapper with await func(...)).

Example

# Runtime: Python 3.12
import functools
import logging
import time

log = logging.getLogger(__name__)

def timed(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            log.info("%s took %.3fs", func.__name__, time.perf_counter() - start)
    return wrapper

def retry(times: int = 3, delay_s: float = 1.0, exceptions: tuple[type[Exception], ...] = (ConnectionError,)):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions:
                    if attempt == times:
                        raise
                    time.sleep(delay_s * attempt)
        return wrapper
    return decorator

@timed
@retry(times=4, delay_s=0.5)
def fetch_rates(day: str) -> dict:
    """Fetch FX rates for a day."""
    return {"USD": 1.0, "EUR": 0.92}

print(fetch_rates.__name__, fetch_rates.__doc__)   # fetch_rates  Fetch FX rates for a day.

Real-world use case

A team wrapped all external calls with a shared @retry decorator that only retries transient error types, plus @timed logging. Retry behaviour became consistent across 30 connectors, and the slow endpoints showed up clearly in the logs.

Common mistakes and red flags

• Forgetting wraps.
• Retrying all exceptions, including bugs.
• Not returning the wrapped function’s result.

Follow-up questions

• Can classes be decorators? — Yes, via __call__; and decorators can decorate classes.
• How do you access the original function? — decorated.__wrapped__ (set by wraps).

Detailed explanation

Protocol

• __enter__(self) runs at the start of the with block. Its return value is bound by as.
• __exit__(self, exc_type, exc, tb) always runs. If it returns True, it suppresses the exception (rarely wanted).
• Async versions are __aenter__/__aexit__ with async with, for example async HTTP sessions (see Concurrent API calls with asyncio).

Common context managers

@contextmanager

@contextmanager
def resource():
    setup()
    try:
        yield value
    finally:
        cleanup()

Without try/finally, cleanup doesn’t run on errors.

Why it matters in pipelines

• Leaked connections exhaust database pools. Leaked file handles hit OS limits in long-running workers.
• Transaction scoping ensures partial writes are rolled back on failure, which is part of idempotency (see Structure of an idempotent, restartable ETL script).
• Cleanup of temp staging files and locks prevents stuck reruns.

Example

# Runtime: Python 3.12
import logging
import shutil
import tempfile
import time
from contextlib import contextmanager
from pathlib import Path

log = logging.getLogger(__name__)

@contextmanager
def staging_dir(prefix: str = "etl_"):
    path = Path(tempfile.mkdtemp(prefix=prefix))
    try:
        yield path
    finally:
        shutil.rmtree(path, ignore_errors=True)

@contextmanager
def step(name: str):
    start = time.perf_counter()
    log.info("start %s", name)
    try:
        yield
        log.info("done %s in %.2fs", name, time.perf_counter() - start)
    except Exception:
        log.exception("failed %s after %.2fs", name, time.perf_counter() - start)
        raise

with step("export orders"), staging_dir() as tmp:
    out = tmp / "orders.csv"
    out.write_text("order_id,amount\n1,10.5\n")
    print(out.exists())        # True; directory removed after block, even on error

Real-world use case

A long-running ingestion service opened database connections without closing them on exceptions, and the pool was exhausted after a few hours of upstream errors. Wrapping connection use in context managers removed the leaks.

Common mistakes and red flags

• @contextmanager without try/finally.
• Swallowing exceptions in __exit__.
• Manual open/close without exception safety.

Follow-up questions

• What does ExitStack solve? — Managing a variable number of context managers (e.g. opening N files) safely.
• Does with conn: close a sqlite3 connection? — No — it commits/rolls back the transaction; close separately.

Detailed explanation

When to use which

• Pydantic:
  • Validating API responses before loading, so you fail fast on contract changes (see What are data contracts, and how do you enforce them?).
  • Config from env and files (pydantic-settings; see Configuration and secrets via environment variables).
  • Request and response models in FastAPI.
  • Record-level validation that quarantines bad rows (see Coding: validate records and split good from bad).
• dataclasses: internal structures passed between your own functions, hot loops creating millions of objects, and libraries avoiding dependencies.
• Neither, for bulk data: millions of rows should live in DataFrames or Arrow, validated with columnar tools such as pandera, Great Expectations or warehouse tests. Per-row Python objects are slow.

Gotchas

• Dataclass mutable defaults need field(default_factory=list) (see Mutable vs immutable types, and the mutable default argument trap).
• Pydantic coercion may hide problems ("123" becomes 123). Use strict mode (ConfigDict(strict=True)) where exact types matter.
• Pydantic v1 → v2 changed the APIs (dict() became model_dump(), validators became field_validator).

Example

# Runtime: Python 3.12 with pydantic 2.x
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, Field, ValidationError, field_validator

@dataclass(slots=True)
class BatchStats:                       # internal, no validation needed
    source: str
    rows: int = 0
    errors: list[str] = field(default_factory=list)

class OrderEvent(BaseModel):            # boundary: validate external payloads
    order_id: str = Field(min_length=1)
    amount: Decimal = Field(ge=0, max_digits=12, decimal_places=2)
    currency: str
    created_at: datetime

    @field_validator("currency")
    @classmethod
    def iso_currency(cls, v: str) -> str:
        v = v.upper()
        if len(v) != 3:
            raise ValueError("currency must be a 3-letter ISO code")
        return v

stats = BatchStats(source="orders_api")
for payload in [{"order_id": "o1", "amount": "19.99", "currency": "usd", "created_at": "2026-03-01T10:00:00Z"},
                {"order_id": "", "amount": "-5", "currency": "EURO", "created_at": "bad"}]:
    try:
        OrderEvent.model_validate(payload); stats.rows += 1
    except ValidationError as e:
        stats.errors.append(f"{payload.get('order_id')!r}: {e.error_count()} errors")
print(stats)   # BatchStats(source='orders_api', rows=1, errors=["'': 4 errors"])

Real-world use case

A partner API silently started returning amount as a formatted string (“1,234.50”). Pydantic validation at ingestion rejected those records with a clear field-level error, instead of loading NULLs or crashing deep inside a transform.

Common mistakes and red flags

• Believing dataclass type hints are enforced.
• Pydantic objects for millions of rows.
• Lax coercion hiding contract breaks.

Follow-up questions

• What’s slots=True? — Uses __slots__ for lower memory and faster attribute access.
• Alternatives? — attrs, msgspec (very fast validation/serialization), TypedDict for typing dicts.

Detailed explanation

Practical guidance

• Dataclasses auto-generate __init__, __repr__ and __eq__. With frozen=True (and the default eq=True) they also get __hash__, and with order=True comparisons. That’s usually better than hand-writing them (see dataclasses vs Pydantic models).
• A good __repr__ pays off in debugging, logs and test failure messages.
• Avoid clever operator overloading that surprises readers.
• Consistency: objects that are equal must have equal hashes.

Example

# Runtime: Python 3.12
from functools import total_ordering
from typing import Iterator

@total_ordering
class Version:
    def __init__(self, text: str):
        self.parts = tuple(int(p) for p in text.split("."))
    def __repr__(self) -> str:
        return f"Version('{'.'.join(map(str, self.parts))}')"
    def __eq__(self, other: object) -> bool:
        return isinstance(other, Version) and self.parts == other.parts
    def __hash__(self) -> int:
        return hash(self.parts)
    def __lt__(self, other: "Version") -> bool:
        return self.parts < other.parts

class Batch:
    def __init__(self, records: list[dict]):
        self._records = records
    def __len__(self) -> int: return len(self._records)
    def __iter__(self) -> Iterator[dict]: return iter(self._records)
    def __bool__(self) -> bool: return bool(self._records)
    def __repr__(self) -> str: return f"Batch(n={len(self)})"

print(sorted({Version("1.10.0"), Version("1.9.3"), Version("1.9.3")}))  # [Version('1.9.3'), Version('1.10.0')]
b = Batch([{"id": 1}])
print(b, len(b), bool(Batch([])))                                       # Batch(n=1) 1 False

sorted compares the integer tuples, so 1.10.0 correctly sorts after 1.9.3. A string sort would get this wrong.

Real-world use case

A schema-migration tool compared version strings as text, so “1.10” sorted before “1.9” and migrations ran out of order. A small Version class with __lt__ and __eq__ fixed the ordering.

Common mistakes and red flags

• __eq__ without a consistent __hash__.
• Hashing mutable fields.
• Useless __repr__ that shows only object addresses.

Follow-up questions

• What happens to __hash__ when you define __eq__? — It’s set to None (unhashable) unless you define it.
• __repr__ vs __str__? — Developer/debug vs end-user display.

Detailed explanation

Why classify

• Retrying permanent errors (401 Unauthorized, 400 Bad Request, schema errors) wastes time and can trigger lockouts or rate limits.
• Not retrying transient errors causes needless failures and pages.
• Clear types make alerts actionable: “schema contract broken” and “provider timeout” call for different responses.

Typical hierarchy

PipelineError
 ├─ RetryableError        (network timeouts, 429, 500/502/503/504, DB deadlock/serialization)
 │    └─ RateLimitedError (carries retry_after seconds)
 └─ PermanentError
      ├─ AuthError        (401/403)
      ├─ ConfigError
      └─ DataValidationError (bad payloads; carries record ids/count)

Design tips

• Inherit from Exception (not BaseException).
• Add attributes (status_code, retry_after, source, batch_id) and a helpful message.
• Translate library exceptions at boundaries (requests.HTTPError → RetryableError or AuthError), so the rest of the code doesn’t depend on library specifics.
• Chain the original with raise ... from e.
• Orchestrator integration: in Airflow, raising AirflowFailException fails a task immediately without retries (for permanent errors), while ordinary exceptions use the configured retries. Similar patterns exist elsewhere (see Sensors vs deferrable operators).
• Tenacity and similar retry libraries can take retry_if_exception_type(RetryableError) (see Retries with exponential backoff and rate limiting).

Alerting tie-in: map exception types to alert routing. AuthError pages the platform team to rotate credentials, DataValidationError notifies the data owner with sample record IDs, and exhausted RetryableErrors open an incident with the provider. Typed errors make that routing a simple lookup instead of fragile log parsing.

Example

# Runtime: Python 3.12 with requests 2.x
import requests

class PipelineError(Exception):
    """Base error for the ingestion pipeline."""

class RetryableError(PipelineError):
    def __init__(self, message: str, *, retry_after: float | None = None):
        super().__init__(message)
        self.retry_after = retry_after

class PermanentError(PipelineError):
    pass

class AuthError(PermanentError):
    pass

RETRYABLE_STATUS = {429, 500, 502, 503, 504}

def get_json(session: requests.Session, url: str, **params) -> dict:
    try:
        resp = session.get(url, params=params, timeout=30)
    except (requests.Timeout, requests.ConnectionError) as e:
        raise RetryableError(f"network error calling {url}") from e

    if resp.status_code in RETRYABLE_STATUS:
        retry_after = resp.headers.get("Retry-After")
        raise RetryableError(f"{resp.status_code} from {url}",
                             retry_after=float(retry_after) if retry_after and retry_after.isdigit() else None)
    if resp.status_code in (401, 403):
        raise AuthError(f"{resp.status_code} from {url}: check credentials/scopes")
    if resp.status_code >= 400:
        raise PermanentError(f"{resp.status_code} from {url}: {resp.text[:200]}")
    return resp.json()

Real-world use case

An API connector retried 401 errors 10 times with backoff, which locked the service account after too many failed authentications. Classifying 401/403 as AuthError (no retry, page immediately) and 429/5xx as retryable fixed both the lockouts and the noisy alerts.

Common mistakes and red flags

• Retrying everything.
• String-matching error messages instead of using types.
• Custom exceptions with no context.

Follow-up questions

• Why translate library exceptions? — Decouples business logic from HTTP/DB library details and centralises classification.
• How to fail an Airflow task without retry? — Raise AirflowFailException.

Detailed explanation

Key APIs

Patterns

• Per-item error handling: submit plus as_completed, then try: f.result() except .... Record failures without stopping other work, and fail the job at the end if any failed (see Coding: validate records and split good from bad).
• Bounded memory with huge inputs: don’t submit a million tasks at once. Submit in windows (keep at most N futures in flight), or batch items (see Coding: batch an iterable into chunks).
• Choosing max_workers: threads for I/O, roughly 4–32 depending on latency and API limits. Processes for CPU, about os.cpu_count().
• Timeouts: use per-call timeouts inside functions (HTTP timeouts). Executor timeouts don’t kill running threads.

Cautions

• Exceptions in futures are silent until you call result() or exception(). Always check them.
• Process pools need picklable functions (top-level defs, not lambdas) and a __main__ guard.
• Thread pools inside web servers or Lambda: watch the lifecycle and resource limits.

Example

# Runtime: Python 3.12
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed

log = logging.getLogger(__name__)

def load_table(table: str) -> int:
    if table == "legacy_events":
        raise RuntimeError("source view missing")
    return {"orders": 1200, "customers": 300}.get(table, 0)

tables = ["orders", "customers", "legacy_events", "refunds"]
results, failures = {}, {}

with ThreadPoolExecutor(max_workers=4) as pool:
    futures = {pool.submit(load_table, t): t for t in tables}
    for fut in as_completed(futures):
        table = futures[fut]
        try:
            results[table] = fut.result()
        except Exception as exc:               # record and continue with other tables
            failures[table] = repr(exc)
            log.error("load failed for %s: %s", table, exc)

print(results, failures)
if failures:
    raise SystemExit(f"{len(failures)} table(s) failed: {sorted(failures)}")

The broad except Exception is deliberate here. Each table’s failure is recorded with full detail, and the job still fails overall at the end, so nothing is swallowed.

Real-world use case

A loader for 120 tables ran sequentially for 2 hours, and one failure stopped all the remaining tables. With an 8-worker pool and per-table error collection, it finished in 20 minutes and reported every failure in one run.

Common mistakes and red flags

• Never calling result(), so errors are lost.
• Submitting millions of futures.
• Lambdas with process pools.

Follow-up questions

• map vs as_completed? — Ordered results vs fastest-first handling and per-item errors.
• Can you cancel running tasks? — Only pending ones; running threads can’t be killed safely.

Detailed explanation

loc vs iloc

Assignment pitfalls

• Chained indexing (df[df.a > 0]["b"] = 1) may modify a temporary copy, which raises SettingWithCopyWarning in older modes. Use df.loc[mask, "b"] = 1.
• Copy-on-Write (opt-in in pandas 2.x and the default in pandas 3.0) makes chained assignment never modify the original, and makes copies lazy. Write code that’s correct under CoW.

Vectorisation ladder (fastest to slowest)

1. Built-in vectorised ops: arithmetic, comparisons, .str.*, .dt.*, np.where, np.select, .map(dict), clip, cut.
2. groupby().agg/transform with built-in aggregations (see pandas merge, concat and groupby/agg).
3. apply on columns with vectorised inner functions.
4. df.apply(func, axis=1), itertuples (faster than iterrows), plain Python loops.
5. iterrows (slowest, and it loses dtypes).

Types matter: keep numeric columns numeric (not object). Use category for low-cardinality strings, and nullable dtypes (Int64, string, boolean) or pd.ArrowDtype for NULL-aware, efficient columns (see Processing data larger than memory in pandas (chunks, dtypes)).

Example

# Runtime: Python 3.12 with pandas 2.2, numpy 2.x
import numpy as np
import pandas as pd

df = pd.DataFrame({
    "order_id": [1, 2, 3, 4],
    "amount": [120.0, 45.0, 600.0, 15.0],
    "country": ["IN", "US", "US", "UK"],
    "ordered_at": pd.to_datetime(["2026-03-01 10:00", "2026-03-01 23:30", "2026-03-02 08:15", "2026-03-02 12:00"]),
}).set_index("order_id")

print(df.loc[2:3, ["amount", "country"]])     # labels 2 AND 3 (inclusive)
print(df.iloc[0:2, 0])                         # positions 0,1 of first column

# Slow: row-wise apply
df["tier_slow"] = df.apply(lambda r: "high" if r["amount"] >= 500 else "mid" if r["amount"] >= 100 else "low", axis=1)

# Fast: vectorised
df["tier"] = np.select([df["amount"] >= 500, df["amount"] >= 100], ["high", "mid"], default="low")
df["order_date"] = df["ordered_at"].dt.date
df.loc[df["country"] == "US", "amount_with_tax"] = df["amount"] * 1.08
print(df[["tier", "tier_slow", "amount_with_tax"]])

Both tier columns are identical: high for 600, mid for 120, low for 45 and 15. Non-US rows get NaN for amount_with_tax.

Real-world use case

A feature-engineering script used df.apply(axis=1) on 8 million rows to compute a tiering rule, which took 14 minutes. np.select did it in 0.4 seconds.

Common mistakes and red flags

• iterrows loops for transformations.
• Chained assignment.
• Forgetting that loc slices are inclusive.

Follow-up questions

• When is apply acceptable? — Small data, or complex logic with no vectorised alternative; consider Polars expressions or Numba otherwise.
• at/iat? — Fast scalar access by label/position.

Detailed explanation

merge

• on, left_on/right_on, left_index/right_index, suffixes=("_l", "_r").
• validate="many_to_one" raises MergeError if the right side has duplicate keys. This prevents silent row multiplication, the pandas equivalent of the SQL fan-out bug (see Gaps and islands: longest streak of consecutive login days).
• indicator=True adds a _merge column (left_only, right_only, both), which is great for reconciliation.
• Key dtypes must match (int vs str keys don’t join). NaN keys match each other in pandas merges, unlike SQL NULLs (see Find the Nth highest salary (several approaches)).

concat

• Vertical: pd.concat(frames, ignore_index=True). Columns are aligned by name, and missing ones are filled with NaN.
• Performance: collect frames in a list and concat once. Don’t concat inside a loop, which is quadratic.
• keys= adds a hierarchical index identifying each source.

groupby

• Named aggregation: df.groupby("country", as_index=False).agg(revenue=("amount", "sum"), orders=("order_id", "nunique")).
• transform("sum") broadcasts group totals back to rows (share of total, group-wise normalisation).
• dropna=False keeps NaN groups (dropped by default!).
• observed=True for categorical keys avoids empty category combinations (the default becomes True in newer pandas).
• sort=False for speed when order doesn’t matter.
• Custom lambdas in agg are slow, so prefer the built-in names.

Example

# Runtime: Python 3.12 with pandas 2.2
import pandas as pd

orders = pd.DataFrame({"order_id": [1, 2, 3, 4], "customer_id": [10, 11, 10, 12],
                       "amount": [100.0, 50.0, 25.0, 80.0]})
customers = pd.DataFrame({"customer_id": [10, 11, 13], "segment": ["SMB", "ENT", "SMB"]})

joined = pd.merge(orders, customers, on="customer_id", how="left",
                  validate="many_to_one", indicator=True)
unmatched = joined.loc[joined["_merge"] == "left_only", "order_id"].tolist()     # [4]

summary = (joined.groupby("segment", dropna=False, as_index=False)
                 .agg(revenue=("amount", "sum"), orders=("order_id", "nunique")))
joined["share_of_customer_spend"] = joined["amount"] / joined.groupby("customer_id")["amount"].transform("sum")

march = pd.concat([orders.assign(month="2026-02"), orders.assign(month="2026-03")], ignore_index=True)
print(unmatched, summary, joined[["order_id", "share_of_customer_spend"]], len(march), sep="\n")

The summary has SMB revenue 125.0 (2 orders), ENT 50.0 (1 order), and a NaN segment with 80.0 (1 order), which is kept by dropna=False. Customer 10’s order shares are 0.8 and 0.2.

Real-world use case

A revenue report doubled one region’s numbers after a dimension file gained duplicate customer rows. Adding validate="many_to_one" to the merge made the job fail immediately with a clear error the next time the file had duplicates.

Common mistakes and red flags

• Merging without validate.
• Concat inside a loop.
• Not realising groupby drops NaN keys by default.

Follow-up questions

• join vs merge? — df.join merges on index by default; merge is more general.
• How do you do an anti-join in pandas? — Merge with indicator=True and keep left_only.

Detailed explanation

Reduce what you load

• usecols=[...] and columns=[...] for Parquet.
• Filter at read time: Parquet filters=[("event_date", "=", "2026-03-01")] with pyarrow, for partition and row-group pruning (see Parquet internals: row groups, column chunks, pages and statistics).
• nrows for sampling during development.

Shrink dtypes

df.memory_usage(deep=True) shows the real usage. Object string columns are often 5–10 times larger than they need to be.

Chunked processing

• pd.read_csv(path, chunksize=500_000, dtype=..., usecols=...) returns an iterator of DataFrames.
• Aggregate each chunk, then combine the partial results (sums and counts combine easily; for means, keep sum and count).
• For per-key operations that need all the rows of a key, sort or partition files by key first, or use a database or engine.
• For Parquet: pyarrow.parquet.ParquetFile(path).iter_batches(batch_size=..., columns=...).

When to switch tools

• Many GBs, joins between large datasets, or repeated jobs: Polars lazy and streaming (scan_parquet(...).collect(engine="streaming")) or DuckDB SQL on files, both of which spill to disk and use all cores.
• Bigger still, or a shared platform: Spark or the warehouse (see Spark vs pandas vs Polars: when to use each, pandas vs Polars vs PySpark).

Example

# Runtime: Python 3.12 with pandas 2.2
import pandas as pd

dtypes = {"store_id": "int32", "country": "category", "status": "category", "amount": "float64"}
totals: dict[tuple[str, str], list[float]] = {}

for chunk in pd.read_csv("sales_2026_q1.csv", usecols=list(dtypes) + ["sold_at"],
                         dtype=dtypes, parse_dates=["sold_at"], chunksize=1_000_000):
    chunk = chunk[chunk["status"] == "completed"]
    chunk["month"] = chunk["sold_at"].dt.strftime("%Y-%m")
    part = chunk.groupby(["country", "month"], observed=True)["amount"].agg(["sum", "count"])
    for key, row in part.iterrows():                     # small partial result
        acc = totals.setdefault(key, [0.0, 0])
        acc[0] += row["sum"]; acc[1] += int(row["count"])

result = (pd.DataFrame([(c, m, s, n, s / n) for (c, m), (s, n) in totals.items()],
                       columns=["country", "month", "revenue", "orders", "avg_order"])
            .sort_values(["country", "month"]))
print(result.head())

iterrows is acceptable here because it loops over the tiny per-chunk aggregate (a few hundred rows), not the raw data.

Real-world use case

A 25 GB CSV crashed a 16 GB container on read_csv. With usecols, category dtypes and 1-million-row chunks, the job ran in 3 GB of RAM. It was later migrated to DuckDB, which runs the whole aggregation as one SQL query in under 2 minutes.

Common mistakes and red flags

• Loading all columns as object dtype.
• Averaging chunk averages.
• Insisting on pandas for data far bigger than RAM.

Follow-up questions

• How do you measure real memory? — df.memory_usage(deep=True).sum() and process RSS.
• Why category dtype? — Stores integer codes plus a small dictionary of unique values.

In depth: How would you process a 50 GB CSV file on a machine with 8 GB of RAM?

Detailed explanation

Options by data shape

Design tips

• Yield batches (thousands of records) for downstream efficiency: bulk inserts, Parquet row groups.
• Track line numbers or offsets for error messages and restartability (resume from offset N).
• Handle bad lines without stopping: yield errors to a quarantine path (see Coding: validate records and split good from bad).
• Avoid f.readlines() and f.read() on big files, both of which load everything.
• For random access or parallelism, split the file into byte ranges aligned to newlines, and process the ranges in separate processes.

Memory check: peak memory should depend on the batch size, not the file size. A quick test is to run the job on a 10× larger file and confirm RSS stays roughly flat.

Example

# Runtime: Python 3.12
import gzip
import json
from pathlib import Path
from typing import Iterator

def read_jsonl_batches(path: str | Path, batch_size: int = 10_000) -> Iterator[tuple[list[dict], list[tuple[int, str]]]]:
    """Yield (good_records, bad_lines) batches with constant memory."""
    path = Path(path)
    opener = gzip.open if path.suffix == ".gz" else open
    good: list[dict] = []
    bad: list[tuple[int, str]] = []
    with opener(path, "rt", encoding="utf-8") as f:
        for lineno, line in enumerate(f, start=1):
            if not line.strip():
                continue
            try:
                good.append(json.loads(line))
            except json.JSONDecodeError:
                bad.append((lineno, line[:500]))
            if len(good) >= batch_size:
                yield good, bad
                good, bad = [], []
    if good or bad:
        yield good, bad

total = 0
for records, errors in read_jsonl_batches("events-2026-03-01.jsonl.gz", batch_size=5_000):
    total += len(records)                 # e.g. write batch to Parquet / bulk insert
    for lineno, snippet in errors:
        print(f"bad line {lineno}: {snippet[:80]!r}")
print("records:", total)

Real-world use case

A support tool parsed a 12 GB gzipped audit log on a 2 GB container by streaming it with batches of 5,000. Bad lines with line numbers went to a quarantine file, so the vendor could fix their exporter.

Common mistakes and red flags

• readlines() on large files.
• Splitting CSV by newline manually.
• Losing line numbers for errors.

Follow-up questions

• How do you resume after failure? — Persist the last processed line/offset or batch number and skip to it on restart.
• Is for line in f buffered? — Yes; Python reads in buffered blocks and splits lines.

Detailed explanation

Levels

Setup

• Libraries and modules: only log = logging.getLogger(__name__), with no basicConfig inside libraries.
• Entry point: logging.basicConfig(level=..., format=...) or logging.config.dictConfig(...).
• Orchestrators (Airflow, Databricks, Lambda) capture stdout and stderr and the root logger. Follow their conventions.

Structured logging

• JSON lines with fields like timestamp, level, logger, message, run_id, task, source, rows, duration_ms can be searched and aggregated in Datadog, CloudWatch, Splunk or Elastic.
• Use extra={...}, a LoggerAdapter, or libraries like structlog or python-json-logger.
• Correlate across systems with a run or trace ID (OpenTelemetry for traces and metrics; see Write-Audit-Publish and blue-green deployments for data).

What to log per step: inputs (source, partition, watermark), outputs (rows read, written, rejected), duration, retries, and the final status. These are the numbers you need during an incident (see Write-Audit-Publish and blue-green deployments for data, Git branching strategies for data teams).

Hygiene

• Lazy formatting: log.info("loaded %s rows", n), not f-strings. It avoids formatting cost when the level is disabled, and keeps messages groupable.
• No secrets (tokens, passwords, connection strings) and minimal PII. Mask emails and IDs if needed.
• Avoid logging every row. Log aggregates and samples.
• Make sure stack traces appear once (log at the boundary that handles the error).

Example

# Runtime: Python 3.12
import json
import logging
import sys
import time
import uuid

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {"ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), "level": record.levelname,
                   "logger": record.name, "msg": record.getMessage()}
        for key in ("run_id", "step", "rows", "duration_ms", "source"):
            if hasattr(record, key):
                payload[key] = getattr(record, key)
        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])

run_id = str(uuid.uuid4())
log = logging.LoggerAdapter(logging.getLogger("orders_etl"), {"run_id": run_id})

start = time.perf_counter()
rows = 12_345
log.info("loaded orders", extra={"step": "load", "rows": rows, "source": "s3://acme-landing/orders/2026-03-01/",
                                  "duration_ms": round((time.perf_counter() - start) * 1000)})

LoggerAdapter merges its own extra and, by default, ignores a call’s extra on Python versions before 3.13. On 3.13+, set merge_extra=True on the adapter so per-call fields like rows are kept. On older versions, pass the per-call fields through a custom adapter.

Real-world use case

During an incident, logs only said “done” for each step, so nobody could tell which partition had loaded zero rows. Structured logs with run_id, step, rows and source let the team find it with one query, and they added an alert on rows == 0.

Common mistakes and red flags

• print in production.
• f-strings everywhere in log calls.
• Logging credentials.
• No row counts.

Follow-up questions

• How do you change log level without code changes? — Env var/config read at startup (LOG_LEVEL).
• Logs vs metrics? — Logs for detail and debugging; metrics for aggregated time series and alerting.

Detailed explanation

Fixtures

• Defined in test files or a shared conftest.py.
• Scopes: function (default, fresh each test), class, module, session (for expensive objects like a SparkSession).
• Teardown through yield: code after yield runs after the test.
• Built-ins: tmp_path (a per-test temp directory), monkeypatch (set env vars and attributes), caplog (captured logs), capsys.
• Fixtures can depend on other fixtures.

Parametrize

• @pytest.mark.parametrize("raw,expected", [("[email protected] ", "[email protected]"), ...]).
• ids= for readable test names. pytest.param(..., marks=pytest.mark.xfail) for known issues.
• Stack decorators for the cartesian product of parameter sets.

What to test in data engineering

• Pure transformation functions: parsing, cleaning, dedupe, business rules. Make them pure, so they’re easy to test (see Building generator pipelines with yield from, OOP: inheritance vs composition).
• Edge cases: NULLs and empties, duplicates, timezone boundaries (DST), late data, unicode, huge values.
• Contracts: output schema (columns and dtypes), uniqueness of keys.
• Integration points: with mocks or fakes (see Mocking external services in tests), or containers (testcontainers for Postgres).
• Spark: a session-scoped local SparkSession fixture, small DataFrames, and chispa or assertDataFrameEqual (PySpark 3.5+) for comparisons.

Organisation: tests/unit/, tests/integration/, and markers (@pytest.mark.integration) to select test types in CI. Use pytest -x -q locally and the --cov coverage plugin.

Example

# Runtime: Python 3.12 with pytest 8.x, pandas 2.2
# src/etl/clean.py
import pandas as pd

def normalise_email(raw: str | None) -> str | None:
    if raw is None or not raw.strip():
        return None
    return raw.strip().lower()

def dedupe_latest(df: pd.DataFrame, key: str, ts: str) -> pd.DataFrame:
    return df.sort_values([key, ts]).drop_duplicates(subset=[key], keep="last").reset_index(drop=True)

# tests/unit/test_clean.py
import pandas as pd
import pytest
from etl.clean import dedupe_latest, normalise_email

@pytest.fixture
def customer_updates() -> pd.DataFrame:
    return pd.DataFrame({"customer_id": [1, 1, 2],
                         "updated_at": pd.to_datetime(["2026-03-01", "2026-03-02", "2026-03-01"]),
                         "segment": ["SMB", "MID", "ENT"]})

@pytest.mark.parametrize("raw,expected", [
    (" [email protected] ", "[email protected]"),
    ("", None),
    ("   ", None),
    (None, None),
], ids=["mixed-case-whitespace", "empty", "blank", "none"])
def test_normalise_email(raw, expected):
    assert normalise_email(raw) == expected

def test_dedupe_latest_keeps_newest(customer_updates):
    out = dedupe_latest(customer_updates, "customer_id", "updated_at")
    assert out.set_index("customer_id")["segment"].to_dict() == {1: "MID", 2: "ENT"}

Real-world use case

A timezone conversion bug only appeared on DST transition days. A parametrized test covering the spring-forward and fall-back dates for three time zones caught it before release, and the cases became part of the regression suite.

Common mistakes and red flags

• Tests that hit real production systems.
• One giant test per module.
• Session-scoped fixtures with mutable shared state.

Follow-up questions

• How do you share fixtures across files? — conftest.py.
• How do you test Spark code fast? — Session-scoped local SparkSession with minimal configs and tiny DataFrames.

Detailed explanation

Concepts

Tooling landscape

• pip + venv: built in. pip-tools (pip-compile) to generate pinned requirements.
• uv: very fast installs and resolution. Commands such as uv init, uv add, uv lock, uv sync and uv run, plus Python version management (uv python install). Also uv pip ... for pip-compatible flows.
• Poetry: dependency management, locking and publishing with its own CLI.
• conda / mamba: useful for non-Python binary dependencies (GDAL, CUDA). Common in data science.
• pipx / uv tool: isolated installs of CLI tools (dbt, sqlfluff).

Practices for data platforms

• Pin dependencies for deployable jobs (Airflow images, Lambda layers, Databricks wheels). Use ranges for libraries published to others.
• Match target runtime versions: Python version, and libraries preinstalled in Databricks Runtime or Snowpark (Anaconda channel) to avoid conflicts (see Databricks Runtime versions and LTS).
• Separate dev dependencies (pytest, ruff, mypy) with dependency groups.
• Build in CI, and run vulnerability scans (pip-audit, Dependabot or Renovate).
• Containers: multi-stage builds that install from the lock file for small, reproducible images.

Example

# pyproject.toml
[project]
name = "acme-etl"
version = "0.4.0"
requires-python = ">=3.11"
dependencies = [
  "pandas>=2.2,<3",
  "pyarrow>=17",
  "requests>=2.32,<3",
  "pydantic-settings>=2.4,<3",
]

[dependency-groups]
dev = ["pytest>=8", "ruff>=0.6", "mypy>=1.11", "responses>=0.25", "moto[s3]>=5"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.ruff]
line-length = 110

uv sync                      # create .venv and install from uv.lock (creates lock if missing)
uv run pytest -q             # run inside the project environment
uv add "duckdb>=1.1"         # add dependency and update lock
uv build                     # build sdist + wheel into dist/

Real-world use case

Airflow tasks failed after a rebuild because an unpinned transitive dependency released a breaking version. Moving to pyproject.toml with uv.lock, and building images from the lock, made rebuilds reproducible, and upgrades now happen through reviewed Renovate PRs.

Common mistakes and red flags

• Installing into system Python.
• Unpinned production dependencies.
• Committing .venv.

Follow-up questions

• Why wheels for Spark jobs? — Packaged code + metadata installable on clusters; consistent across nodes.
• requirements.txt vs pyproject.toml? — Requirements is a flat install list; pyproject defines the project and its dependency specs (locks capture exact versions).

Detailed explanation

Modern syntax (3.10+ / 3.12+)

• Built-in generics: list[int], dict[str, float], tuple[int, ...].
• Unions: int | None (instead of Optional[int]).
• typing helpers: Any, Literal["dev", "prod"], TypedDict (typed dict shapes), Protocol (structural interfaces; see OOP: inheritance vs composition), Callable[[int], str], Iterator[T], Final, TypeAlias.
• Python 3.12 generics syntax: def first[T](xs: list[T]) -> T: ... and type JSON = dict[str, "JSON"] | list["JSON"] | str | int | float | bool | None.
• from __future__ import annotations for postponed evaluation (forward references) on older versions.

Checkers

Adoption strategy

• Start with public functions and data structures (dataclasses, TypedDicts).
• Enable checks gradually (--strict per module), and use # type: ignore[code] sparingly with reasons.
• Run the checker in CI and pre-commit, alongside a linter (ruff).
• Use stubs for untyped libraries (types-requests, pandas-stubs, boto3-stubs).

Where types help most in DE code

• None-handling (str | None forces checks before .lower()).
• Config objects and records passed between stages.
• Refactors across large codebases.
• Clear signatures for shared libraries used by many pipelines.

Limits: DataFrame column types aren’t captured by standard hints. Use pandera or Pydantic for schema validation, and Polars or pandas typed schemas where available.

Example

# Runtime: Python 3.12 (check with: mypy --strict etl_types.py)
from dataclasses import dataclass
from datetime import date
from typing import Iterable, Iterator, Literal, TypedDict

type Env = Literal["dev", "ci", "prod"]

class RawOrder(TypedDict):
    order_id: str
    amount: str
    coupon: str | None

@dataclass(frozen=True, slots=True)
class Order:
    order_id: str
    amount_cents: int
    coupon: str | None
    load_date: date

def parse_orders(rows: Iterable[RawOrder], load_date: date) -> Iterator[Order]:
    for r in rows:
        coupon = r["coupon"].strip().upper() if r["coupon"] else None   # None-safe; mypy enforces
        yield Order(r["order_id"], round(float(r["amount"]) * 100), coupon, load_date)

def target_schema(env: Env) -> str:
    return {"dev": "dbt_dev", "ci": "dbt_ci", "prod": "analytics"}[env]

orders = list(parse_orders([{"order_id": "o1", "amount": "12.30", "coupon": None}], date(2026, 3, 1)))
print(orders, target_schema("prod"))

Calling target_schema("staging") is flagged by the type checker before it can fail at runtime with a KeyError. In real money pipelines, use Decimal rather than float to compute cents.

Real-world use case

A refactor changed a function to return str | None, and three call sites crashed in production on .upper() of None. After mypy was added to CI, a similar change was flagged at PR time with exact line numbers.

Common mistakes and red flags

• Any everywhere.
• Believing hints validate at runtime.
• Not running a checker at all.

Follow-up questions

• TypedDict vs dataclass? — TypedDict types plain dicts (e.g. JSON); dataclass creates real objects.
• How do you type a decorator? — ParamSpec and TypeVar (Callable[P, R]).

Detailed explanation

Clarify first

1. Lists: index positions, keep as JSON or arrays, or explode into child rows (one row per item)? For analytics, exploding into a child table is usually right (see explode, flattening nested JSON and arrays).
2. Separator: . or _. What if keys already contain the separator?
3. Empty dicts and lists: drop them, or keep them as empty values?
4. Types: keep native types, or stringify everything?
5. Depth: very deep nesting can hit the recursion limit (default about 1000). An iterative stack avoids that.

Algorithm (dict and index flattening)

• flatten(obj, prefix):
  • If obj is a dict: for each k, v, recurse with prefix + sep + k.
  • If a list (with indexing): for each i, v, recurse with prefix + sep + str(i).
  • Otherwise: out[prefix] = obj.
• Time is O(total number of nodes), and output size equals the number of leaves.

Explode variant: produce several rows, a parent row of scalar fields plus a list of child rows with a parent key. This matches relational modelling.

Tools: pandas.json_normalize(data, record_path=[...], meta=[...], sep=".") for record lists with metadata. Spark explode and struct selection at scale (see explode, flattening nested JSON and arrays).

Example

# Runtime: Python 3.12
from typing import Any

def flatten(obj: Any, sep: str = ".", index_lists: bool = True) -> dict[str, Any]:
    """Iteratively flatten nested dicts/lists into a single-level dict."""
    out: dict[str, Any] = {}
    stack: list[tuple[str, Any]] = [("", obj)]
    while stack:
        prefix, value = stack.pop()
        if isinstance(value, dict) and value:
            for k, v in value.items():
                stack.append((f"{prefix}{sep}{k}" if prefix else str(k), v))
        elif isinstance(value, list) and value and index_lists:
            for i, v in enumerate(value):
                stack.append((f"{prefix}{sep}{i}" if prefix else str(i), v))
        else:
            if prefix in out:
                raise ValueError(f"key collision on '{prefix}'")
            out[prefix] = value
    return dict(sorted(out.items()))

event = {
    "event_id": "e1",
    "user": {"id": 42, "geo": {"country": "IN", "city": "Pune"}},
    "items": [{"sku": "A", "qty": 2}, {"sku": "B", "qty": 1}],
    "tags": [],
}
print(flatten(event))
# {'event_id': 'e1', 'items.0.qty': 2, 'items.0.sku': 'A', 'items.1.qty': 1, 'items.1.sku': 'B',
#  'tags': [], 'user.geo.city': 'Pune', 'user.geo.country': 'IN', 'user.id': 42}

Empty containers are kept as leaves. The collision check catches inputs like {"a.b": 1, "a": {"b": 2}}, which would otherwise overwrite silently.

Real-world use case

A webhook ingestion stored flattened event payloads for a schemaless audit table, with arrays kept as JSON strings, and exploded items into a separate child table for analytics. The collision check caught a vendor that sent both meta.source and a nested meta object.

Common mistakes and red flags

• Not asking how lists should be handled.
• Recursion without considering depth.
• Silent key collisions.

Follow-up questions

• How would you unflatten? — Split keys by separator and build nested dicts (numeric segments → lists if required).
• How would you do this for 100 GB of JSON? — Spark/Polars with explicit schemas and explode, not Python loops.

Detailed explanation

Clarify

• What defines “latest”: an event timestamp, an update timestamp, or a sequence number? How are ties broken?
• NULL or missing timestamps: treat them as oldest, or reject?
• Is the input sorted? If so, you can stream with constant memory per key group.
• Do deleted records (tombstones) remove the key, or stay as the latest state?
• Memory constraints: what if there are too many unique keys to hold in memory?

Algorithm (dict)

• best = {}. For each record r: k = r[key]. If k isn’t in best or (r.ts, r.seq) > (best[k].ts, best[k].seq), set best[k] = r.
• Comparing tuples gives deterministic tiebreaking.
• Output: best.values(), optionally sorted.

Variants

• Sorted input (by key): use itertools.groupby, and take the max per group, with O(1) memory per group.
• Too many keys for memory: external sort by key (split, sort and merge), a database, DuckDB, or Spark.
• Streaming dedupe of exact duplicates by event ID: a bounded time-window set or a Bloom filter (accepting false positives), like watermark-based dedupe in streaming (see Watermarks and late data in Structured Streaming).

Complexity: O(n) time, O(u) space for u unique keys. A sort-based approach is O(n log n).

Example

# Runtime: Python 3.12
from datetime import datetime
from operator import itemgetter
from typing import Iterable

def latest_by_key(records: Iterable[dict], key: str = "customer_id",
                  ts: str = "updated_at", seq: str = "ingest_seq") -> list[dict]:
    best: dict = {}
    for r in records:
        if r.get(ts) is None:
            continue                                   # policy: skip records without timestamp
        k = r[key]
        rank = (r[ts], r.get(seq, 0))
        current = best.get(k)
        if current is None or rank > (current[ts], current.get(seq, 0)):
            best[k] = r
    return sorted(best.values(), key=itemgetter(key))

updates = [
    {"customer_id": 1, "segment": "SMB", "updated_at": datetime(2026, 3, 1, 9), "ingest_seq": 10},
    {"customer_id": 1, "segment": "MID", "updated_at": datetime(2026, 3, 2, 9), "ingest_seq": 11},
    {"customer_id": 2, "segment": "ENT", "updated_at": datetime(2026, 3, 1, 9), "ingest_seq": 12},
    {"customer_id": 2, "segment": "MID", "updated_at": datetime(2026, 3, 1, 9), "ingest_seq": 13},  # tie
    {"customer_id": 3, "segment": "SMB", "updated_at": None, "ingest_seq": 14},
]
for r in latest_by_key(updates):
    print(r["customer_id"], r["segment"])
# 1 MID
# 2 MID   (tie on updated_at, higher ingest_seq wins)