Quick summary
Summarize this blog with AI
This interview guide uses one complete extract–validate–load exercise instead of placeholder orchestration code. You will load a bounded CSV batch into an in-memory SQLite database, make retries idempotent by event identity, and reject conflicting replays without partially committing the batch. Python 3.12.14 and its standard library are sufficient.
The fixture is deliberately small. It teaches data contracts and transaction semantics; it does not claim production throughput, distributed exactly-once delivery, or compatibility with an untested Airflow release. Give each question two practice points: one for a working result and one for explaining its boundary.
1. What contract should ingestion enforce?
Prompt: Accept exactly event_id, product, and amount_cents. IDs and product names must be nonempty, IDs must be unique within the batch, and amount must be a nonnegative integer no larger than 10 billion cents. Reject malformed records before opening a load transaction. A batch is at most 1 MiB and 1000 records.
import csv
import io
import re
import sqlite3
COLUMNS = ["event_id", "product", "amount_cents"]
def parse_batch(text):
if not isinstance(text, str):
raise TypeError("CSV input must be text")
if len(text.encode("utf-8")) > 1024 * 1024:
raise ValueError("batch exceeds byte limit")
reader = csv.DictReader(io.StringIO(text, newline=""), strict=True)
if reader.fieldnames != COLUMNS:
raise ValueError("unexpected CSV columns")
rows, seen = [], set()
for line, row in enumerate(reader, start=2):
if len(rows) == 1000:
raise ValueError("batch exceeds row limit")
if None in row or any(value is None for value in row.values()):
raise ValueError(f"wrong field count at record {line}")
event, product, raw = (row[key] for key in COLUMNS)
if not event.strip() or not product.strip() or len(event) > 100 or len(product) > 100:
raise ValueError(f"invalid identity at record {line}")
if event in seen:
raise ValueError("duplicate event ID in batch")
if not re.fullmatch(r"[0-9]{1,11}", raw) or int(raw) > 10_000_000_000:
raise ValueError(f"invalid amount at record {line}")
seen.add(event)
rows.append((event, product, int(raw)))
return rows
batch = "event_id,product,amount_cents\ne1,book,1200\ne2,game,700\n"
print(parse_batch(batch))
Expected output:
[('e1', 'book', 1200), ('e2', 'game', 700)]
Integer cents avoid binary floating-point rounding for this currency model. This contract permits zero and rejects negative amounts; refunds would need an explicit signed-event design. The row limit bounds this tutorial's retained list, while the byte cap bounds the supplied input. For huge feeds, stream into a staged table and preserve an equivalent validation/commit boundary.
2. How can replay be idempotent without hiding conflicts?
Prompt: Replaying the same event and payload should insert nothing. Reusing an event ID with different data must fail. The function owns its transaction and rejects an already-open caller transaction.
def load_batch(connection, text):
rows = parse_batch(text)
if connection.in_transaction:
raise ValueError("loader requires no active transaction")
inserted = 0
with connection:
connection.execute("BEGIN IMMEDIATE")
for event, product, amount in rows:
old = connection.execute(
"SELECT product, amount_cents FROM events WHERE event_id = ?", (event,)
).fetchone()
if old is not None:
if old != (product, amount):
raise ValueError(f"conflicting replay for {event}")
continue
connection.execute("INSERT INTO events VALUES (?, ?, ?)",
(event, product, amount))
inserted += 1
return inserted
connection = sqlite3.connect(":memory:")
connection.execute("""CREATE TABLE events (
event_id TEXT PRIMARY KEY, product TEXT NOT NULL,
amount_cents INTEGER NOT NULL CHECK(amount_cents >= 0))""")
print(load_batch(connection, batch))
print(load_batch(connection, batch))
print(connection.execute("SELECT COUNT(*), SUM(amount_cents) FROM events").fetchone())
Expected output:
2
0
(2, 1900)
A primary key enforces event identity. Matching existing payloads are recognized as replays; conflicting payloads are rejected. Bound parameters keep values separate from SQL. BEGIN IMMEDIATE obtains SQLite's write transaction before checking existing records. A warehouse or server database needs its own concurrency-safe upsert/constraint strategy; copying a check-then-insert pattern without that protection can race.
3. What happens if a conflict follows a new event?
Prompt: The next batch adds e3, then conflicts with e1. Prove the first insert does not survive the failed transaction.
conflict = "event_id,product,amount_cents\ne3,pen,100\ne1,book,9999\n"
try:
load_batch(connection, conflict)
except ValueError as exc:
print(str(exc))
print(connection.execute("SELECT event_id FROM events ORDER BY event_id").fetchall())
print(connection.in_transaction)
Expected output:
conflicting replay for e1
[('e1',), ('e2',)]
False
The context manager rolls back when an exception escapes. Catching the conflict inside the transaction and pretending success would break that guarantee. Parsing errors likewise propagate before loading begins. A database transaction cannot roll back an email, external API call, or file write already performed elsewhere; use an outbox or separately designed side-effect protocol when those are required.
4. What should retry?
Prompt: Classify a malformed amount, a conflicting replay, and a transient database-unavailable failure.
Answer: malformed data and identity conflicts need correction or quarantine, not automatic retries. A transient availability problem may justify bounded retries with backoff and an operation identity, after confirming the transaction outcome. A lost acknowledgment can leave the caller unsure whether a commit happened; replaying the same stable event IDs is why idempotency matters. Do not claim that a scheduler retry means exactly-once execution.
5. What does orchestration add?
Prompt: Describe how a scheduler would run this loader without inventing a pass-filled DAG.
Answer: a scheduled task supplies a fixed source batch identity and content, calls the tested loader, records the outcome, and exposes failure for review. Its connection credentials belong in managed configuration. Pin the orchestrator and provider versions and test their supported API before writing deployment-specific task definitions. Retry policy, concurrency, backfills, late data, and alert routing are separate choices; a Python function alone does not implement them.
6. What outputs prove the small pipeline worked?
print(connection.execute("""SELECT product, SUM(amount_cents)
FROM events GROUP BY product ORDER BY product""").fetchall())
print(load_batch(connection, "event_id,product,amount_cents\n"))
try:
load_batch(connection, "event_id,product,amount_cents\ne4,book,-5\n")
except ValueError:
print("invalid batch rejected")
print(connection.execute("SELECT COUNT(*) FROM events").fetchone()[0])
connection.close()
Expected output:
[('book', 1200), ('game', 700)]
0
invalid batch rejected
2
A header-only batch is a valid no-op under this contract. The negative amount changes nothing. The normal result, replay result, conflict rollback, and invalid-batch result together prove more than a single “success” print. For a real feed, reconcile source count, accepted count, rejected count, freshness, and stable event identity before optimizing batch size.
References: CSV parsing, SQLite connection context managers, SQLite transaction modes, and parameter binding.
