Module 07 // Integration

Python + SQL

> Bridge declarative SQL with imperative Python. Cursors, parameterized queries, transactions, ORMs, and dataframes.

sqlite3 — Standard Library

Built into Python — perfect for embedded apps, prototypes, and tests. Always use parameter substitution to prevent SQL injection.

import sqlite3

with sqlite3.connect(class="str">"app.db") as conn:
    conn.execute(class="str">""class="str">"
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            email TEXT UNIQUE NOT NULL
        )
    "class="str">"")

    class=class="str">"com"># SAFE — parameterized
    conn.execute(class="str">"INSERT INTO users (email) VALUES (?)", (class="str">"ada@example.com",))

    class=class="str">"com"># NEVER do this — SQL injection!
    class=class="str">"com"># conn.execute(fclass="str">"INSERT INTO users (email) VALUES ('{user_input}')")

    rows = conn.execute(class="str">"SELECT * FROM users WHERE id = ?", (1,)).fetchall()
    for row in rows:
        print(row)

psycopg — PostgreSQL

import psycopg

with psycopg.connect(class="str">"dbname=app user=postgres") as conn:
    with conn.cursor() as cur:
        cur.execute(
            class="str">"SELECT name, salary FROM employees WHERE department = %s",
            (class="str">"Engineering",),
        )
        for name, salary in cur:
            print(name, salary)

SQLAlchemy ORM

Map tables to classes; SQLAlchemy generates the SQL for you.

from sqlalchemy import create_engine, String, Integer
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase): pass

class Employee(Base):
    __tablename__ = class="str">"employees"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    salary: Mapped[int]

engine = create_engine(class="str">"postgresql+psycopg://localhost/app")

with Session(engine) as session:
    new = Employee(name=class="str">"Ada", salary=95000)
    session.add(new)
    session.commit()

    class=class="str">"com"># Query
    high_earners = session.query(Employee).filter(Employee.salary > 80000).all()

pandas — DataFrames from SQL

Pull query results straight into a pandas DataFrame for analysis.

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine(class="str">"postgresql+psycopg://localhost/app")

df = pd.read_sql(
    class="str">""class="str">"
    SELECT department, salary
    FROM employees
    WHERE hired_at >= '2020-01-01'
    "class="str">"",
    engine,
)

class=class="str">"com"># Aggregate in pandas
summary = df.groupby(class="str">"department")[class="str">"salary"].agg([class="str">"mean", class="str">"count", class="str">"max"])
print(summary)

class=class="str">"com"># Write back
summary.to_sql(class="str">"dept_summary", engine, if_exists=class="str">"replace")

Best Practices

  • Always parameterize — never f-string user input into SQL.
  • Use transactions — wrap related writes in with blocks.
  • Pool connections in long-running apps (SQLAlchemy does this by default).
  • Stream large results with cursors or generators instead of fetchall().
  • Index your hot columns — see the indexing article.