Cohort 4Register

PART 6 · PYTHON FOR DATA ENGINEERING · CHAPTER 32

Python and databases: parameters, transactions, upserts

FreeLesson 69 of 78

sqlite3 ships with Python, so every example here runs anywhere. The same calls work with psycopg (PostgreSQL), mysql-connector and the Snowflake connector.

BULK LOAD, THEN QUERY WITH PARAMETERSPYTHON 3.12
import csv, sqlite3

con = sqlite3.connect(":memory:")
con.execute("""CREATE TABLE orders (order_id INTEGER PRIMARY KEY,
    customer_id INT, order_date TEXT, status TEXT, amount REAL)""")
rows = [tuple(r.values()) for r in csv.DictReader(open("orders.csv"))]
con.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", rows)
con.commit()

q = "SELECT COUNT(*), SUM(amount) FROM orders WHERE status = ?"
print(con.execute(q, ("delivered",)).fetchone())

user_input = "x' OR '1'='1"
unsafe = f"SELECT COUNT(*) FROM orders WHERE status = '{user_input}'"
print("f-string:", con.execute(unsafe).fetchone()[0], "rows")
safe = "SELECT COUNT(*) FROM orders WHERE status = ?"
print("param:   ", con.execute(safe, (user_input,)).fetchone()[0], "rows")

Output

(12, 18972.0)
f-string: 15 rows
param:    0 rows
TRANSACTIONS AND UPSERTSPYTHON 3.12
with con:                              # commit on success
    con.execute("INSERT INTO orders VALUES "
                "(116, 8, '2025-05-02', 'delivered', 499)")

try:
    with con:                          # rollback on any error
        con.execute("UPDATE orders SET status = 'returned' "
                    "WHERE order_id = 101")
        con.execute("INSERT INTO orders VALUES "
                    "(101, 1, '2025-01-10', 'dup', 0)")    # PK clash
except sqlite3.IntegrityError as e:
    print("rolled back:", e)
q101 = "SELECT status FROM orders WHERE order_id = 101"
print(con.execute(q101).fetchone())

con.execute("""INSERT INTO orders
    VALUES (115, 5, '2025-04-28', 'returned', 1598)
    ON CONFLICT(order_id) DO UPDATE SET status = excluded.status""")
print(con.execute("SELECT order_id, status FROM orders "
                  "WHERE order_id IN (115, 116)").fetchall())

Output

rolled back: UNIQUE constraint failed: orders.order_id
('delivered',)
[(115, 'returned'), (116, 'delivered')]
PANDAS IN AND OUT OF SQLPYTHON 3.12
import pandas as pd

pd.read_csv("products.csv").to_sql("products", con, index=False)
q = ("SELECT category, COUNT(*) AS n, MAX(price) AS top "
     "FROM products GROUP BY category")
print(pd.read_sql(q, con))

Output

      category  n   top
0  Electronics  4  1999
1         Home  2  1199
2   Stationery  2   249

For large results use cursor.fetchmany(10_000) in a loop, or pd.read_sql(..., chunksize=), so the result never has to fit in memory at once.

Practice before you move on

  1. Load products.csv into SQLite and query the average price for one category passed as a parameter.
  2. Write an upsert that changes the status of one order and check it with a SELECT.