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