claims = Table("claims", metadata,
    Column("claim_id", String(10), primary_key=True),
    Column("policy_id", String(10),
           ForeignKey("policies.policy_id"), nullable=False),
    Column("vehicle_id", String(10),
           ForeignKey("vehicles.vehicle_id"),
           nullable=False, index=True),
    Column("coverage", String(4), nullable=False),
    Column("accident_date", Date, nullable=False),
    Column("report_date", Date, nullable=False),
    Column("status", String(10), nullable=False),
    Column("paid", Float, nullable=False),
    Column("outstanding", Float, nullable=False),
    CheckConstraint("report_date >= accident_date",
                    name="reported_after_accident"),
    CheckConstraint("paid >= 0 AND outstanding >= 0",
                    name="amounts_nonnegative"),
)
metadata.create_all(engine)       # CREATE TABLE ... for all six

ORDER = [persons, policies, vehicles,          # parents
         policy_drivers, coverages, claims]    # first

with engine.begin() as conn:          # commit at the end,
    for table in ORDER:               # roll back on any error
        for chunk in pd.read_csv(RAW / f"{table.name}.csv",
                                 chunksize=20_000,
                                 parse_dates=DATES[table.name]):
            rows = (chunk.astype(object)
                    .where(chunk.notna(), None)   # NaN -> NULL
                    .to_dict(orient="records"))
            conn.execute(insert(table), rows)

SELECT strftime('%Y', p.inception_date) AS policy_year,
       COUNT(*) AS vehicles,
       SUM(MAX(0, julianday(MIN(p.expiry_date,
                   COALESCE(p.cancellation_date, p.expiry_date),
                   '2025-12-31'))
                  - julianday(p.inception_date) + 1) / 365.25)
           AS car_years
FROM vehicles v JOIN policies p USING (policy_id)
GROUP BY policy_year;

evil = "x' OR '1'='1"          # typed into a search box

# WRONG: the input becomes part of the SQL
conn.execute(text(f"SELECT * FROM policies "
                  f"WHERE policy_id = '{evil}'"))
# ... WHERE policy_id = 'x' OR '1'='1'   -> 78,062 rows

# RIGHT: a bound parameter is only ever a value
conn.execute(text("SELECT * FROM policies "
                  "WHERE policy_id = :pid"), {"pid": evil})
#                                          -> 0 rows

SELECT * FROM claims WHERE vehicle_id = 'VEH000005';

SEARCH claims USING INDEX ix_claims_vehicle_id
       (vehicle_id=?)

q = "SELECT coverage, annual_premium FROM coverages"

# (a) everything at once
pd.read_sql(text(q), conn).groupby("coverage").sum()

# (b) 10,000 rows at a time, streamed
parts = [chunk.groupby("coverage")["annual_premium"].sum()
         for chunk in pd.read_sql(
             text(q), conn.execution_options(stream_results=True),
             chunksize=10_000)]
pd.concat(parts).groupby(level=0).sum()

# (c) the database aggregates; 3 rows come back
pd.read_sql(text("SELECT coverage, SUM(annual_premium) "
                 "FROM coverages GROUP BY coverage"), conn)

class Base(DeclarativeBase):
    pass

class QuoteRecord(Base):
    __tablename__ = "quotes"
    id: Mapped[int] = mapped_column(primary_key=True)
    created_at: Mapped[datetime] = mapped_column(
        UTCDateTime, default=lambda: datetime.now(UTC))
    request: Mapped[dict[str, Any]] = mapped_column(JSON)
    liab_premium: Mapped[float]
    coll_premium: Mapped[float | None]
    total_premium: Mapped[float]
    model_version: Mapped[str] = mapped_column(String(40),
                                               index=True)

with Session(engine) as s:
    s.add(QuoteRecord(request=..., liab_premium=4648.59, ...))
    s.commit()
    rec = s.get(QuoteRecord, 1)

E  Differing items:
E  {'created_at': '2026-09-30T14:58:24.157592'} !=
E  {'created_at': '2026-09-30T14:58:24.157592Z'}

class UTCDateTime(TypeDecorator[datetime]):
    """Stored in UTC; always returned time-zone aware."""
    impl = DateTime(timezone=True)
    cache_ok = True

    def process_bind_param(self, value, dialect):
        if value is not None and value.tzinfo is None:
            raise ValueError("naive datetime: attach a "
                             "time zone before storing")
        return value.astimezone(UTC) if value else None

    def process_result_value(self, value, dialect):
        if value is not None and value.tzinfo is None:
            return value.replace(tzinfo=UTC)
        return value

engine = sm.make_engine(
    "postgresql+psycopg://postgres:***@localhost:5433/postgres")
counts = sm.migrate(engine)      # the same function

What changed: the date arithmetic (julianday becomes date subtraction, MAX/MIN become GREATEST/LEAST). Keep dialect-specific SQL in few, tested places.

PostgreSQL is a server: loading over a network connection is slower than writing a local file, and in exchange you get many concurrent users, permissions, backups, and replication.

• Keys, foreign keys, and constraints are in the schema
• SQLite connections switch foreign keys on
• Loads run in one transaction
• Every value in a query is a bound parameter

• I index the columns I filter and join on
• Large tables are chunked or aggregated in the database
• Timestamps are stored in UTC and read back aware
• The same code runs on SQLite and PostgreSQL