Databases & SQL with Python¶

import sqlite3
from contextlib import contextmanager
from typing import Generator

@contextmanager
def get_connection(path: str = "app.db") -> Generator[sqlite3.Connection, None, None]:
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row   # rows behave like dicts
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# ── Safe: parameterized query — ALWAYS use this pattern ──────────
with get_connection() as conn:
    cur = conn.cursor()
    cur.execute(
        "SELECT id, email FROM users WHERE age > ? AND city = ?",
        (18, "Bengaluru"),
    )
    rows = cur.fetchall()
    for row in rows:
        print(row["id"], row["email"])   # row_factory enables dict access

# ── Bulk insert with executemany ───────────────────────────────────
users = [("u1", "ada@acme.io", 30), ("u2", "grace@acme.io", 25)]
with get_connection() as conn:
    conn.executemany(
        "INSERT INTO users (id, email, age) VALUES (?, ?, ?)",
        users,
    )

# ── DANGER: never build SQL with f-strings ────────────────────────
# user_input = "'; DROP TABLE users; --"
# cur.execute(f"SELECT * FROM users WHERE name = '{user_input}'")
# ^ This executes arbitrary SQL — classic SQL injection vector

from sqlalchemy import (
    create_engine, String, Integer, ForeignKey, Index
)
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship, Session
)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id:         Mapped[int]         = mapped_column(primary_key=True)
    email:      Mapped[str]         = mapped_column(String(255), unique=True, index=True)
    tier:       Mapped[str]         = mapped_column(String(50), default="starter")
    orders:     Mapped[list["Order"]] = relationship(back_populates="user")

class Order(Base):
    __tablename__ = "orders"

    id:         Mapped[int]  = mapped_column(primary_key=True)
    user_id:    Mapped[int]  = mapped_column(ForeignKey("users.id"), index=True)
    product_id: Mapped[str]  = mapped_column(String(100))
    value:      Mapped[float]
    user:       Mapped["User"] = relationship(back_populates="orders")

    # Composite index on commonly-queried columns
    __table_args__ = (
        Index("ix_orders_user_product", "user_id", "product_id"),
    )

# ── CRUD operations ───────────────────────────────────────────────
engine = create_engine("postgresql+psycopg2://user:pass@localhost/acme")
Base.metadata.create_all(engine)

with Session(engine) as session:
    # Create
    user = User(email="ada@acme.io", tier="enterprise")
    session.add(user)
    session.flush()   # assigns user.id without committing

    order = Order(user_id=user.id, product_id="p42", value=299.99)
    session.add(order)
    session.commit()

    # Read — ORM generates parameterized SQL
    enterprise_users = (
        session
        .query(User)
        .filter(User.tier == "enterprise")
        .order_by(User.email)
        .all()
    )   # SQL: SELECT * FROM users WHERE tier = 'enterprise' ORDER BY email

from sqlalchemy.orm import Session, selectinload, joinedload

with Session(engine) as session:
    # ── BAD: N+1 — 1 query for users, then 1 per user for orders ──
    users = session.query(User).all()
    for user in users:
        # Each .orders access fires a new SELECT — 1001 queries for 1000 users!
        total = sum(o.value for o in user.orders)

    # ── GOOD: eager loading — 2 queries total ─────────────────────
    users = (
        session
        .query(User)
        .options(selectinload(User.orders))   # fetch all orders in one IN query
        .all()
    )
    for user in users:
        total = sum(o.value for o in user.orders)   # no additional queries

    # ── ALSO GOOD: JOIN loading — 1 query with a SQL JOIN ─────────
    users = (
        session
        .query(User)
        .options(joinedload(User.orders))
        .all()
    )