In this post we are going to cover Python and databases for data engineers: SQL injection and parameterized queries, bulk loads, transactions and commits, upserts, connection management, and streaming big results in batches. These are 7 real Python interview questions, asked the way they come up on a real project, and the last one is a scenario where we load a daily file into a table so a rerun never duplicates and a failure never leaves half the data.
Almost every pipeline reads from or writes to a database through a Python driver. The SQL is usually fine. The problems are in the Python around it: how values get into the query, when it commits, who closes the connection, and how many rows sit in memory at once. That's exactly what interviewers probe.
Every Python step below runs on the sample data, and the output under each step is the real result.
Part 8 of the Real Time Python for Data Engineers Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's create some sample data
- Question 1: A developer builds SQL with f-strings and user input. What is the risk, and how do parameterized queries fix it?
- Question 2: Inserting one hundred thousand rows one by one takes forever. How do you load rows in bulk?
- Question 3: Half of a load was committed before an error. How do transactions and commits work from Python?
- Question 4: The target table needs new rows inserted and existing rows updated. How do you upsert from Python?
- Question 5: Connections pile up and the database refuses new ones. How do you manage connections properly?
- Question 6: A query result is too big for memory. How do you stream rows from the database in batches?
- Scenario 7: You must load a daily file into a database table safely, so a rerun never duplicates and a failure never leaves half the data. What do you build?
- The full script
The short answer
Pass values as parameters, never with f-strings. Load in bulk with executemany or COPY inside one transaction, upsert with INSERT ... ON CONFLICT DO UPDATE, close connections with contextlib.closing (a sqlite3 with block only commits or rolls back), stream big results with fetchmany, and make daily loads replace their slice in one transaction.
Keep that in mind. Every question below is one piece of it.
Let's create some sample data
We use SQLite through the sqlite3 module, so everything runs from one database file, shop.db, with a small table for each question. The setup also writes the daily orders file for the scenario, and the same day resent by the source with a bad row.
import sqlite3
from contextlib import closing
from pathlib import Path
# One SQLite database file in the current folder, with a small table for each question
with closing(sqlite3.connect("shop.db")) as db:
db.executescript("""
CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL);
INSERT INTO users (id, name) VALUES (1, 'ann'), (2, 'bob'), (3, 'cara');
CREATE TABLE sales (id INTEGER PRIMARY KEY, store TEXT NOT NULL, qty INTEGER NOT NULL);
CREATE TABLE daily_load (id INTEGER PRIMARY KEY, qty INTEGER NOT NULL);
CREATE TABLE products (sku TEXT PRIMARY KEY, price_cents INTEGER NOT NULL, updated_on TEXT NOT NULL);
INSERT INTO products (sku, price_cents, updated_on) VALUES ('A100', 1199, '2026-10-20'), ('B200', 899, '2026-10-20');
CREATE TABLE job_log (id INTEGER PRIMARY KEY, step TEXT NOT NULL);
CREATE TABLE orders (day TEXT NOT NULL, order_id INTEGER NOT NULL,
qty INTEGER NOT NULL CHECK (qty >= 0), PRIMARY KEY (day, order_id));
""")
# Q7: the daily file, and the same day resent by the source with a bad row
Path("orders_1024.csv").write_text("order_id,qty\n1001,5\n1002,7\n1003,2\n", encoding="utf-8")
Path("orders_1024_resent.csv").write_text("order_id,qty\n1001,5\n1002,-7\n1003,2\n", encoding="utf-8")Question 1: A developer builds SQL with f-strings and user input. What is the risk, and how do parameterized queries fix it?
The risk is SQL injection: user input pasted into the SQL text can change the query itself.
Here's the users table:
| id | name |
|---|---|
| 1 | ann |
| 2 | bob |
| 3 | cara |
And a value typed into a form: x, a quote, then OR '1'='1. Let's build the query with an f-string, print it, and run it.
import sqlite3
conn = sqlite3.connect("shop.db", autocommit=False)
name = "x' OR '1'='1" # typed into a web form
sql = f"SELECT id, name FROM users WHERE name = '{name}'"
print(sql)
print(conn.execute(sql).fetchall())SELECT id, name FROM users WHERE name = 'x' OR '1'='1'
[(1, 'ann'), (2, 'bob'), (3, 'cara')]The quote closes the string, and the OR makes the condition true for every row. The input became part of the SQL, and the query returns all three users instead of none.
Now let's pass the same value as a parameter: a ? placeholder, and the value in a tuple. Then a named placeholder, :name with a dictionary.
sql = "SELECT id, name FROM users WHERE name = ?"
print(conn.execute(sql, (name,)).fetchall())
sql = "SELECT id, name FROM users WHERE name = :name"
print(conn.execute(sql, {"name": "ann"}).fetchall())[]
[(1, 'ann')]The driver sends the value as data, not SQL, so the first query returns []. The named placeholder works too, and finds (1, 'ann').
Placeholders bind values, not table or column names. For dynamic names, use an allow list, or psycopg.sql.Identifier in PostgreSQL. And psycopg 3.3, on Python 3.14, turns t-strings into parameterized queries.
Always use placeholders instead of string formatting.
Question 2: Inserting one hundred thousand rows one by one takes forever. How do you load rows in bulk?
The slow version calls execute() in a loop, often with a commit() after every row. That's 100,000 statements and 100,000 commits, and on a remote database, a network round trip for each one.
Use executemany: one statement, many parameter rows, inside one transaction. Feed it from a generator, so the rows never sit in memory as one big list. Let's load 100,000 sales rows and check them with one count and sum query.
rows = ((i, f"S{i % 5}", i % 7) for i in range(1, 100_001))
with conn: # one transaction
conn.executemany("INSERT INTO sales (id, store, qty)"
" VALUES (?, ?, ?)", rows)
cur = conn.execute("SELECT COUNT(*), SUM(qty) FROM sales")
print(cur.fetchone())(100000, 300000)(100000, 300000): 100,000 rows, and a total quantity of 300,000.
For millions of rows, split the input with itertools.batched and commit each batch, so a failure costs one batch, not the whole load. Then make sure a rerun can't double the batches that already committed.
from itertools import batched
for chunk in batched(rows, 10_000):
with conn: # commit per batch
conn.executemany(INSERT_SQL, chunk)In PostgreSQL, the fast path is COPY. With psycopg 3, cursor.copy streams rows into a table, much faster than executemany inserts.
with conn.cursor() as cur:
with cur.copy("COPY stg_sales FROM STDIN") as cp:
for row in rows:
cp.write_row(row)COPY into a staging table, then insert into the target in one transaction.
Question 3: Half of a load was committed before an error. How do transactions and commits work from Python?
A transaction makes a group of statements all or nothing.
The bug: the loader commits after every row. Row 3 has no quantity and breaks the NOT NULL rule. Let's run it.
batch = [(1, 5), (2, 7), (3, None), (4, 2)] # 3: no qty
insert = "INSERT INTO daily_load (id, qty) VALUES (?, ?)"
count = "SELECT COUNT(*) FROM daily_load"
try:
for row in batch:
conn.execute(insert, row)
conn.commit() # commit per row
except sqlite3.IntegrityError as e:
print("failed:", e)
print("rows in table:", conn.execute(count).fetchone()[0])failed: NOT NULL constraint failed: daily_load.qty
rows in table: 2The load stopped with NOT NULL constraint failed: daily_load.qty, and the table has 2 rows, already committed. Half a load.
The fix is one transaction for the whole batch. On a sqlite3 connection, the with conn: block commits when the block succeeds, and rolls back when it raises. Let's reset the table, run the same batch, then fix the bad row and run it again.
with conn:
conn.execute("DELETE FROM daily_load") # reset the demo
try:
with conn: # commit, or roll back on error
conn.executemany(insert, batch)
except sqlite3.IntegrityError as e:
print("rolled back:", e)
print("rows in table:", conn.execute(count).fetchone()[0])
batch[2] = (3, 4) # fix the bad row
with conn:
conn.executemany(insert, batch)
print("rows in table:", conn.execute(count).fetchone()[0])rolled back: NOT NULL constraint failed: daily_load.qty
rows in table: 0
rows in table: 4The insert failed, the whole batch rolled back, and the table has 0 rows. After the fix, all 4 rows land together.
Know your driver:
- We opened the connection with
autocommit=False, the value the docs recommend since Python 3.12. The default is stillLEGACY_TRANSACTION_CONTROL. - The sqlite3
withblock doesn't close the connection. A psycopg 3withblock does.
Commit once, at the end of the batch. Never commit inside the loop, unless the load can restart safely from where it stopped.
Question 4: The target table needs new rows inserted and existing rows updated. How do you upsert from Python?
Let the database do it in one statement: INSERT ... ON CONFLICT on the key, DO UPDATE. It works in SQLite and in PostgreSQL. The products table has two products, A100 and B200.
| sku | price_cents | updated_on |
|---|---|---|
| A100 | 1199 | 2026-10-20 |
| B200 | 899 | 2026-10-20 |
Today's batch changes the price of A100 and adds C300. When the SKU already exists, the statement updates the price and the date instead of inserting. excluded is the row you tried to insert. The WHERE clause is a guard: update only when the incoming row is newer, so a late file can't overwrite fresher data. Let's run it with executemany.
UPSERT = """
INSERT INTO products (sku, price_cents, updated_on)
VALUES (?, ?, ?)
ON CONFLICT (sku) DO UPDATE SET
price_cents = excluded.price_cents,
updated_on = excluded.updated_on
WHERE excluded.updated_on > products.updated_on"""
batch = [("A100", 1299, "2026-10-24"), # changed price
("C300", 450, "2026-10-24")] # new product
with conn:
conn.executemany(UPSERT, batch)This step prints nothing; it prepares the data for the next step.
Now let's look at the table, run the batch again, and compare.
SHOW = "SELECT * FROM products ORDER BY sku"
first = conn.execute(SHOW).fetchall()
print(*first, sep="\n")
with conn:
conn.executemany(UPSERT, batch) # the rerun
again = conn.execute(SHOW).fetchall()
print("same after rerun:", again == first)('A100', 1299, '2026-10-24')
('B200', 899, '2026-10-20')
('C300', 450, '2026-10-24')
same after rerun: TrueThree products now. A100 has the new price, 1299 cents, dated 2026-10-24. B200 is untouched at 899, and C300 is new at 450. The rerun prints same after rerun: True: the table didn't change, so it's safe to rerun.
In PostgreSQL, it's the same statement with %s placeholders.
COPY into staging first, then one INSERT ... SELECT with ON CONFLICT, or a MERGE on PostgreSQL 15 and later.
Question 5: Connections pile up and the database refuses new ones. How do you manage connections properly?
Usually the code opens a connection per call or per row, and an exception skips the close. It goes like this:
- A connection per call, and an error skips
close(). - Each leaked connection holds a slot on the server.
- The database hits its connection limit, and new connections are refused.
The classic trap in sqlite3: with conn: looks like it closes the connection, but it only commits or rolls back. Let's check.
with conn: # commit or rollback, not close
conn.execute("INSERT INTO job_log (step) VALUES (?)",
("bulk load",))
print("still open:", conn.execute("SELECT 1").fetchone())still open: (1,)After the block, the connection still answers a query: still open: (1,).
To close it, wrap the connection in contextlib.closing. The inner with block handles the transaction, and the outer one closes the connection, even when an error is raised. Let's use the connection after the block.
from contextlib import closing
def open_db(path="shop.db"):
return sqlite3.connect(path, autocommit=False)
with closing(open_db()) as db:
with db: # the transaction
cur = db.execute("SELECT COUNT(*) FROM job_log")
print("job_log rows:", cur.fetchone()[0])
db.execute("SELECT 1") # closed after the blockjob_log rows: 1
sqlite3.ProgrammingError: Cannot operate on a closed database.Inside, job_log rows: 1. After the block, sqlite3.ProgrammingError: Cannot operate on a closed database.
In PostgreSQL, a psycopg 3 connection's with block commits and closes. A long-running service should use a pool, for example a SQLAlchemy engine with pool_pre_ping=True.
One connection per job, opened at the start, closed by a with block, and passed to the functions that need it. Never open one inside a loop.
Question 6: A query result is too big for memory. How do you stream rows from the database in batches?
fetchall() pulls every row into one Python list. With a hundred million rows, that list is what kills the job. Pull one batch at a time instead.
| fetchall() | fetchmany(n) or iterate the cursor | |
|---|---|---|
| IN MEMORY | every row at once | one batch at a time |
| 100 MILLION ROWS | the job runs out of memory | the same small batch, again and again |
fetchmany takes a size and returns the next batch, and an empty list at the end. The walrus loop runs while there are rows. Let's read the 100,000 sales rows in batches of 25,000, and keep a running total.
cur = conn.execute("SELECT id, qty FROM sales ORDER BY id")
batches = rows = total = 0
while batch := cur.fetchmany(25_000):
batches += 1
rows += len(batch)
total += sum(qty for _, qty in batch)
print(f"batches={batches} rows={rows} total_qty={total}")batches=4 rows=100000 total_qty=300000batches=4 rows=100000 total_qty=300000, the same as the bulk load.
With pandas, read_sql takes a chunksize and yields one DataFrame per chunk. Let's try it.
import pandas as pd
sql = "SELECT id, qty FROM sales ORDER BY id"
for chunk in pd.read_sql(sql, conn, chunksize=25_000):
print(type(chunk).__name__, len(chunk), chunk.qty.sum())DataFrame 25000 74997
DataFrame 25000 75006
DataFrame 25000 74994
DataFrame 25000 75003Four DataFrames of 25,000 rows each. Write each chunk out, for example to Parquet.
One catch: check that your driver streams, with pandas chunks too. psycopg's default cursor pulls the whole result into memory, so use a named, server-side cursor there.
Then a restart can resume after the last key.
Scenario 7: You must load a daily file into a database table safely, so a rerun never duplicates and a failure never leaves half the data. What do you build?
The design: parse the whole file first, then replace that day's slice of the table inside one transaction. Either everything changes, or nothing does.
- Parse the whole file, so bad values fail before the database.
- One transaction: delete the day, insert its rows.
- Commit or roll back, never half a day.
read_rows parses the whole CSV first, so a bad value fails before the database is touched.
import csv
def read_rows(path, day):
with open(path, newline="", encoding="utf-8") as f:
return [(day, int(r["order_id"]), int(r["qty"]))
for r in csv.DictReader(f)]This step prints nothing; it prepares the data for the next step.
Then, inside one with conn: block, load_day deletes the rows for that day and inserts the new ones with executemany.
DELETE = "DELETE FROM orders WHERE day = ?"
INSERT = ("INSERT INTO orders (day, order_id, qty)"
" VALUES (?, ?, ?)")
def load_day(conn, path, day):
rows = read_rows(path, day) # parse first
with conn: # all or nothing
conn.execute(DELETE, (day,))
conn.executemany(INSERT, rows)
return len(rows)This step prints nothing; it prepares the data for the next step.
Let's run it, rerun it, and then load the resent file, which has a negative quantity.
day, path = "2026-10-24", "orders_1024.csv"
COUNT = "SELECT COUNT(*) FROM orders"
print("run loaded", load_day(conn, path, day))
print("rerun loaded", load_day(conn, path, day))
print("table has", conn.execute(COUNT).fetchone()[0])
try:
load_day(conn, "orders_1024_resent.csv", day)
except sqlite3.IntegrityError as e:
print("failed run:", e)
print("table still has", conn.execute(COUNT).fetchone()[0])run loaded 3
rerun loaded 3
table has 3
failed run: CHECK constraint failed: qty >= 0
table still has 3The first run loaded 3 rows. The rerun loaded the same 3, and the table has 3, not 6. The resent file breaks the check constraint, CHECK constraint failed: qty >= 0. The insert failed, the delete rolled back with it, and the table still has the 3 good rows. Readers never see an empty day.
In production I add three things:
- In PostgreSQL, COPY into staging, then delete the day and insert from staging, in one transaction.
- A load log table, with the file name and row count.
- A totals check that fails the run when the counts don't match the file.
Parse first, then delete and insert in one transaction. A rerun replaces the day, and a failure changes nothing.
Key points to remember
- Parameters (? or :name), never f-strings with user input
- Bulk: executemany in one transaction; COPY in PostgreSQL
- with conn commits or rolls back; closing() closes
- Upsert: INSERT ... ON CONFLICT DO UPDATE, safe to rerun
- Stream with fetchmany; replace a day in one transaction
The full script
Here is everything in one script: the sample data and every step from this post, in order. Copy it and run it top to bottom.
Show the full Python script (sample data and every step)
# ========================================================================
# Python Interview Q&A #8 - Python and databases
# TechBrothersIT. Sample data is invented for teaching.
# Every step below was run with Python 3.14.6 and pandas 3.1.0; the outputs in the video and the blog
# post are the real outputs. Run it in an empty folder: it writes small sample files.
# ========================================================================
# Step 0: sample data
import sqlite3
from contextlib import closing
from pathlib import Path
# One SQLite database file in the current folder, with a small table for each question
with closing(sqlite3.connect("shop.db")) as db:
db.executescript("""
CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL);
INSERT INTO users (id, name) VALUES (1, 'ann'), (2, 'bob'), (3, 'cara');
CREATE TABLE sales (id INTEGER PRIMARY KEY, store TEXT NOT NULL, qty INTEGER NOT NULL);
CREATE TABLE daily_load (id INTEGER PRIMARY KEY, qty INTEGER NOT NULL);
CREATE TABLE products (sku TEXT PRIMARY KEY, price_cents INTEGER NOT NULL, updated_on TEXT NOT NULL);
INSERT INTO products (sku, price_cents, updated_on) VALUES ('A100', 1199, '2026-10-20'), ('B200', 899, '2026-10-20');
CREATE TABLE job_log (id INTEGER PRIMARY KEY, step TEXT NOT NULL);
CREATE TABLE orders (day TEXT NOT NULL, order_id INTEGER NOT NULL,
qty INTEGER NOT NULL CHECK (qty >= 0), PRIMARY KEY (day, order_id));
""")
# Q7: the daily file, and the same day resent by the source with a bad row
Path("orders_1024.csv").write_text("order_id,qty\n1001,5\n1002,7\n1003,2\n", encoding="utf-8")
Path("orders_1024_resent.csv").write_text("order_id,qty\n1001,5\n1002,-7\n1003,2\n", encoding="utf-8")
# Step 1: SQL built with an f-string: the input changes the query
import sqlite3
conn = sqlite3.connect("shop.db", autocommit=False)
name = "x' OR '1'='1" # typed into a web form
sql = f"SELECT id, name FROM users WHERE name = '{name}'"
print(sql)
print(conn.execute(sql).fetchall())
# Step 2: The same input as a parameter: ? and :name placeholders
sql = "SELECT id, name FROM users WHERE name = ?"
print(conn.execute(sql, (name,)).fetchall())
sql = "SELECT id, name FROM users WHERE name = :name"
print(conn.execute(sql, {"name": "ann"}).fetchall())
# Step 3: Bulk load: executemany from a generator, in one transaction
rows = ((i, f"S{i % 5}", i % 7) for i in range(1, 100_001))
with conn: # one transaction
conn.executemany("INSERT INTO sales (id, store, qty)"
" VALUES (?, ?, ?)", rows)
cur = conn.execute("SELECT COUNT(*), SUM(qty) FROM sales")
print(cur.fetchone())
# Step 4: The bug: commit after every row, then an error at row 3
batch = [(1, 5), (2, 7), (3, None), (4, 2)] # 3: no qty
insert = "INSERT INTO daily_load (id, qty) VALUES (?, ?)"
count = "SELECT COUNT(*) FROM daily_load"
try:
for row in batch:
conn.execute(insert, row)
conn.commit() # commit per row
except sqlite3.IntegrityError as e:
print("failed:", e)
print("rows in table:", conn.execute(count).fetchone()[0])
# Step 5: One transaction: with conn rolls back the whole batch on error
with conn:
conn.execute("DELETE FROM daily_load") # reset the demo
try:
with conn: # commit, or roll back on error
conn.executemany(insert, batch)
except sqlite3.IntegrityError as e:
print("rolled back:", e)
print("rows in table:", conn.execute(count).fetchone()[0])
batch[2] = (3, 4) # fix the bad row
with conn:
conn.executemany(insert, batch)
print("rows in table:", conn.execute(count).fetchone()[0])
# Step 6: Upsert: INSERT ... ON CONFLICT DO UPDATE, only when the row is newer
UPSERT = """
INSERT INTO products (sku, price_cents, updated_on)
VALUES (?, ?, ?)
ON CONFLICT (sku) DO UPDATE SET
price_cents = excluded.price_cents,
updated_on = excluded.updated_on
WHERE excluded.updated_on > products.updated_on"""
batch = [("A100", 1299, "2026-10-24"), # changed price
("C300", 450, "2026-10-24")] # new product
with conn:
conn.executemany(UPSERT, batch)
# Step 7: Check the result, then rerun: nothing changes
SHOW = "SELECT * FROM products ORDER BY sku"
first = conn.execute(SHOW).fetchall()
print(*first, sep="\n")
with conn:
conn.executemany(UPSERT, batch) # the rerun
again = conn.execute(SHOW).fetchall()
print("same after rerun:", again == first)
# Step 8: with conn ends the transaction but does not close the connection
with conn: # commit or rollback, not close
conn.execute("INSERT INTO job_log (step) VALUES (?)",
("bulk load",))
print("still open:", conn.execute("SELECT 1").fetchone())
# Step 9: contextlib.closing closes the connection when the block ends
# This step is EXPECTED TO FAIL: it shows the error the video explains.
try:
from contextlib import closing
def open_db(path="shop.db"):
return sqlite3.connect(path, autocommit=False)
with closing(open_db()) as db:
with db: # the transaction
cur = db.execute("SELECT COUNT(*) FROM job_log")
print("job_log rows:", cur.fetchone()[0])
db.execute("SELECT 1") # closed after the block
except Exception as e:
print(type(e).__name__ + ":", e)
# Step 10: Stream a big result in batches with fetchmany
cur = conn.execute("SELECT id, qty FROM sales ORDER BY id")
batches = rows = total = 0
while batch := cur.fetchmany(25_000):
batches += 1
rows += len(batch)
total += sum(qty for _, qty in batch)
print(f"batches={batches} rows={rows} total_qty={total}")
# Step 11: pandas: read_sql with chunksize yields one DataFrame per chunk
import pandas as pd
sql = "SELECT id, qty FROM sales ORDER BY id"
for chunk in pd.read_sql(sql, conn, chunksize=25_000):
print(type(chunk).__name__, len(chunk), chunk.qty.sum())
# Step 12: Parse the whole daily file first
import csv
def read_rows(path, day):
with open(path, newline="", encoding="utf-8") as f:
return [(day, int(r["order_id"]), int(r["qty"]))
for r in csv.DictReader(f)]
# Step 13: Load one day: delete and insert the day in one transaction
DELETE = "DELETE FROM orders WHERE day = ?"
INSERT = ("INSERT INTO orders (day, order_id, qty)"
" VALUES (?, ?, ?)")
def load_day(conn, path, day):
rows = read_rows(path, day) # parse first
with conn: # all or nothing
conn.execute(DELETE, (day,))
conn.executemany(INSERT, rows)
return len(rows)
# Step 14: Run, rerun, and a failed resend: the table never doubles or half-loads
day, path = "2026-10-24", "orders_1024.csv"
COUNT = "SELECT COUNT(*) FROM orders"
print("run loaded", load_day(conn, path, day))
print("rerun loaded", load_day(conn, path, day))
print("table has", conn.execute(COUNT).fetchone()[0])
try:
load_day(conn, "orders_1024_resent.csv", day)
except sqlite3.IntegrityError as e:
print("failed run:", e)
print("table still has", conn.execute(COUNT).fetchone()[0])
Part 9: Testing, Packaging and Code Quality
The sample data on this page is invented for teaching. Every output is the real output of the code above it, run with Python 3.14.6 and pandas 3.1.0. Part of the Real Time Python for Data Engineers Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Python and library documentation for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.