Data Engineer Interview Questions — Python

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.

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.

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.

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.

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.

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.

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.

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.

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 ifs 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.

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.

# 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.

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.

# 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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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,<3"]

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

pyproject.toml replaced setup.py as the standard configuration, and any of hatchling, setuptools or flit works as the backend. setup.py still works and is no longer what you reach for first.

The src/ layout matters more than it looks. Without it, the package directory sits next to your tests and Python finds it on the path whether or not it is installed, so your tests pass against the working tree and the actual installed package is never exercised. With src/, tests can only import what was really installed, which catches a missing file in the package data before your users do.

During development, install it editable:

pip install -e .

The environment points at your working tree, so edits take effect without reinstalling. A normal install copies files into site-packages and your changes do nothing until you reinstall, which is a confusing hour for anyone who has not met it.

Version pinning splits by role. A library declares ranges, because pinning exactly forces conflicts on anyone installing it alongside something else. An application pins exactly in a lock file, because reproducibility is the point. Getting this backwards produces either a library nobody can install or a deployment that drifts.

A console entry point turns a function into a command:

[project.scripts]
run-ingest = "pipelines_common.cli:main"

For internal distribution, a private index or a Git URL in the requirements file both work without publishing anything publicly.

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.

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.

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.

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:

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.

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.

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 <= 0:
            raise StopIteration
        self.n -= 1
        return self.n + 1

Making it self-consuming like this is a design choice with a cost: Countdown(3) can only be looped once. The better pattern for a reusable collection is to keep the state out of the container and return a fresh iterator, or simply make __iter__ a generator function, which gives you a new one per call for free.

Two details for follow-ups. StopIteration is caught by the for loop itself, so you never see it unless you call next() manually; pass a default to next(it, None) to avoid it. And iter() falls back to __getitem__ with integer indices from zero for classes predating the protocol, which is why some old sequence classes are iterable without defining __iter__.

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.

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.

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.

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.

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.

The three formats answer different questions.

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.

# 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.

# 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.

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.

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.

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.

# 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.

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.

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.

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.

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.

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.