{ "cells": [ { "cell_type": "markdown", "id": "cell-000", "metadata": {}, "source": [ "# Lesson 6 — Enterprise Data Stores: SQL & SQLAlchemy\n", "\n", "**Companion notebook** for the Lesson 6 slides. You'll migrate the motor book from CSV\n", "files into a relational database with keys and constraints, load it in one transaction,\n", "answer the exposure and claims questions in SQL, see constraints and transactions\n", "protect the data, avoid SQL injection, stream a large table in chunks, and use the ORM.\n", "\n", "Everything runs on SQLite (a single file, built into Python). The same code runs on\n", "PostgreSQL by changing the URL." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-001", "metadata": {}, "outputs": [], "source": [ "def check(label, got, expected, tol=1e-6):\n", " \"\"\"Compare your answer with the expected value; prints ✅ or ❌ (never raises).\"\"\"\n", " import math\n", " try:\n", " if isinstance(expected, float) or isinstance(got, float):\n", " ok = math.isclose(float(got), float(expected), rel_tol=tol, abs_tol=tol)\n", " elif isinstance(expected, (list, tuple)) and expected and isinstance(expected[0], str):\n", " ok = list(got) == list(expected)\n", " elif hasattr(expected, \"__len__\") and not isinstance(expected, str):\n", " import numpy as np\n", " ok = np.allclose(np.asarray(got, dtype=float), np.asarray(expected, dtype=float), rtol=tol, atol=tol)\n", " else:\n", " ok = got == expected\n", " except Exception:\n", " ok = False\n", " print((\"✅ \" if ok else \"❌ \") + f\"{label}: got {got!r}, expected {expected!r}\")\n", "import tracemalloc\n", "from datetime import date, datetime, timezone\n", "from pathlib import Path\n", "\n", "import pandas as pd\n", "from sqlalchemy import (CheckConstraint, Column, Date, Float, ForeignKey, Integer, MetaData, String, Table,\n", " create_engine, event, func, insert, select, text)\n", "from sqlalchemy.exc import IntegrityError\n", "from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column\n", "\n", "RAW = Path(\"../data/motor/raw\")\n", "DB = Path(\"motor.db\")\n", "DB.unlink(missing_ok=True)\n", "engine = create_engine(f\"sqlite:///{DB}\")\n", "\n", "\n", "@event.listens_for(engine, \"connect\")\n", "def _foreign_keys_on(dbapi_conn, _):\n", " dbapi_conn.execute(\"PRAGMA foreign_keys = ON\") # SQLite ignores foreign keys unless asked" ] }, { "cell_type": "markdown", "id": "cell-002", "metadata": {}, "source": [ "## 1. The schema says what the CSV pipeline only assumed" ] }, { "cell_type": "code", "execution_count": null, "id": "cell-003", "metadata": {}, "outputs": [], "source": [ "metadata = MetaData()\n", "persons = Table(\"persons\", metadata,\n", " Column(\"person_id\", String(10), primary_key=True),\n", " Column(\"date_of_birth\", Date, nullable=False),\n", " Column(\"gender\", String(1), nullable=False),\n", " Column(\"licence_date\", Date, nullable=False))\n", "policies = Table(\"policies\", metadata,\n", " Column(\"policy_id\", String(10), primary_key=True),\n", " Column(\"policyholder_id\", String(10), ForeignKey(\"persons.person_id\"), nullable=False),\n", " Column(\"renewal_of\", String(10), ForeignKey(\"policies.policy_id\")),\n", " Column(\"inception_date\", Date, nullable=False),\n", " Column(\"expiry_date\", Date, nullable=False),\n", " Column(\"cancellation_date\", Date),\n", " Column(\"channel\", String(12), nullable=False),\n", " Column(\"latitude\", Float, nullable=False),\n", " Column(\"longitude\", Float, nullable=False),\n", " Column(\"written_premium\", Float, nullable=False),\n", " CheckConstraint(\"expiry_date >= inception_date\", name=\"term_positive\"))\n", "vehicles = Table(\"vehicles\", metadata,\n", " Column(\"vehicle_id\", String(10), primary_key=True),\n", " Column(\"policy_id\", String(10), ForeignKey(\"policies.policy_id\"), nullable=False, index=True),\n", " Column(\"make\", String(20), nullable=False),\n", " Column(\"model\", String(20), nullable=False),\n", " Column(\"vehicle_style\", String(12), nullable=False),\n", " Column(\"model_year\", Integer, nullable=False),\n", " Column(\"vehicle_value\", Float, nullable=False),\n", " Column(\"annual_mileage\", Float, nullable=False),\n", " CheckConstraint(\"annual_mileage > 0\", name=\"mileage_positive\"))\n", "coverages = Table(\"coverages\", metadata,\n", " Column(\"policy_id\", String(10), ForeignKey(\"policies.policy_id\"), nullable=False),\n", " Column(\"vehicle_id\", String(10), ForeignKey(\"vehicles.vehicle_id\"), primary_key=True),\n", " Column(\"coverage\", String(4), primary_key=True),\n", " Column(\"deductible\", Integer, nullable=False),\n", " Column(\"limit\", Float, nullable=False),\n", " Column(\"annual_premium\", Float, nullable=False))" ] }, { "cell_type": "markdown", "id": "cell-004", "metadata": {}, "source": [ "---\n", "## Your turn 1 — the claims table\n", "\n", "Columns (all required except none): `claim_id` (primary key), `policy_id` (foreign key\n", "to policies), `vehicle_id` (foreign key to vehicles, indexed), `coverage`,\n", "`accident_date`, `report_date`, `status`, `paid`, `outstanding`. Constraints:\n", "- `report_date >= accident_date`, named `reported_after_accident`\n", "- `paid >= 0 AND outstanding >= 0`, named `amounts_nonnegative`\n", "- `coverage IN ('LIAB', 'COLL', 'COMP')`, named `coverage_code`" ] }, { "cell_type": "code", "execution_count": null, "id": "cell-005", "metadata": {}, "outputs": [], "source": [ "claims = None\n", "... # your code here\n", "metadata.create_all(engine)\n", "with engine.connect() as conn:\n", " print(conn.execute(text(\"SELECT sql FROM sqlite_master WHERE name = 'claims'\")).scalar())" ] }, { "cell_type": "markdown", "id": "cell-006", "metadata": {}, "source": [ "## Your turn 2 — load it, all or nothing\n", "\n", "Write `load(table)`: read `RAW / f\"{table.name}.csv\"` in chunks of 20,000 rows, convert\n", "the table's `Date` columns to Python dates (`DATES[table.name]` lists them), turn NaN\n", "into `None`, and `conn.execute(insert(table), rows)`. Run all tables **inside one**\n", "`with engine.begin() as conn:` block, parents before children, so that a failure\n", "anywhere leaves the database empty." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-007", "metadata": {}, "outputs": [], "source": [ "ORDER = [t for t in [persons, policies, vehicles, coverages, claims] if t is not None]\n", "DATES = {t.name: [c.name for c in t.columns if isinstance(c.type, Date)] for t in ORDER}\n", "counts = None\n", "... # your code here\n", "check(\"rows loaded\", None if counts is None else sum(counts.values()), 438426)\n", "print(f\"{DB.stat().st_size / 1e6:.1f} MB\")" ] }, { "cell_type": "markdown", "id": "cell-008", "metadata": {}, "source": [ "## Your turn 3 — ask the database\n", "\n", "In SQL, not pandas:\n", "1. car-years earned to 31 December 2025, summed over vehicles: for each vehicle, the\n", " days from its policy's inception to the earliest of expiry, cancellation, and\n", " 2025-12-31, plus one, over 365.25 (never negative). In SQLite, `julianday()` turns a\n", " date into a day number; `MIN(a, b, c)` with several arguments is the smallest;\n", " `COALESCE(x, y)` is x unless x is NULL.\n", "2. liability claims with an accident on or before 2025-12-31: use a **bound parameter**\n", " (`:end`), not string formatting." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-009", "metadata": {}, "outputs": [], "source": [ "car_years = n_liab = None\n", "... # your code here\n", "check(\"car-years\", None if car_years is None else round(car_years, 1), 72780.5)\n", "check(\"liability claims\", n_liab, 4565)" ] }, { "cell_type": "markdown", "id": "cell-010", "metadata": {}, "source": [ "## Your turn 4 — constraints and transactions\n", "\n", "Try to insert each bad claim below **in its own transaction** and count how many the\n", "database rejects (catch `IntegrityError`). Then insert a good claim and an orphan in the\n", "**same** transaction: how many claims are there afterwards?" ] }, { "cell_type": "code", "execution_count": null, "id": "cell-011", "metadata": {}, "outputs": [], "source": [ "base = {\"coverage\": \"LIAB\", \"accident_date\": date(2025, 6, 1), \"report_date\": date(2025, 6, 2),\n", " \"status\": \"Open\", \"paid\": 0.0, \"outstanding\": 1000.0, \"policy_id\": \"POL000005\", \"vehicle_id\": \"VEH000005\"}\n", "BAD = [{**base, \"claim_id\": \"X1\", \"vehicle_id\": \"VEH999999\"}, # no such vehicle\n", " {**base, \"claim_id\": \"X2\", \"report_date\": date(2025, 5, 1)}, # reported before the accident\n", " {**base, \"claim_id\": \"X3\", \"paid\": -50.0}, # negative payment\n", " {**base, \"claim_id\": \"CLM000001\"}, # duplicate key\n", " {**base, \"claim_id\": \"X5\", \"coverage\": \"GAP\"}] # unknown coverage\n", "rejected = after = None\n", "... # your code here\n", "check(\"bad claims rejected\", rejected, 5)\n", "check(\"claims after the failed transaction\", after, 11562)" ] }, { "cell_type": "markdown", "id": "cell-012", "metadata": {}, "source": [ "## Your turn 5 — SQL injection\n", "\n", "A search box passes a policy id to this function. Show what it returns for the input\n", "below, then write `find_policy_safe` with a bound parameter and store how many rows each\n", "returns." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-013", "metadata": {}, "outputs": [], "source": [ "def find_policy_unsafe(policy_id):\n", " with engine.connect() as conn:\n", " return conn.execute(text(f\"SELECT * FROM policies WHERE policy_id = '{policy_id}'\")).all()\n", "\n", "\n", "evil = \"x' OR '1'='1\"\n", "n_unsafe = n_safe = None\n", "... # your code here\n", "check(\"rows returned by the unsafe query\", n_unsafe, 78062)\n", "check(\"rows returned by the safe query\", n_safe, 0)" ] }, { "cell_type": "markdown", "id": "cell-014", "metadata": {}, "source": [ "## Your turn 6 — bigger than memory\n", "\n", "Total annual premium by coverage from the 224,988-row `coverages` table, two ways,\n", "measuring peak memory with `tracemalloc`: (a) `pd.read_sql` of the whole table, then\n", "`groupby`; (b) `pd.read_sql(..., chunksize=10_000)`, a groupby per chunk, then combine.\n", "Then (c) let the database do it: `GROUP BY` in SQL. Store the three results (Series\n", "indexed by coverage, rounded to cents) and the two peaks in bytes." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-015", "metadata": {}, "outputs": [], "source": [ "Q = \"SELECT coverage, annual_premium FROM coverages\"\n", "whole = chunked = in_db = peak_whole = peak_chunked = None\n", "... # your code here\n", "check(\"chunks give the same totals\", None if chunked is None else bool((whole - chunked).abs().max() < 0.01), True)\n", "check(\"the database gives the same totals\", None if in_db is None else bool((whole - in_db).abs().max() < 0.01), True)\n", "check(\"chunks use less memory\", None if peak_chunked is None else peak_chunked < peak_whole / 3, True)" ] }, { "cell_type": "markdown", "id": "cell-016", "metadata": {}, "source": [ "## Your turn 7 — the ORM\n", "\n", "Map a `Quote` class to a `quotes` table: `id` (integer primary key), `created_at`\n", "(datetime), `vehicle_style` (string), `premium` (float). Create the table, add three\n", "quotes in one session (Sports 2,521.59, Sedan 938.09, Sports 1,800.00), commit, then use\n", "`select(func.sum(Quote.premium)).where(Quote.vehicle_style == \"Sports\")` to total the\n", "Sports premium." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-017", "metadata": {}, "outputs": [], "source": [ "class Base(DeclarativeBase):\n", " pass\n", "\n", "\n", "sports_total = None\n", "... # your code here\n", "check(\"Sports premium\", None if sports_total is None else round(sports_total, 2), 4321.59)" ] }, { "cell_type": "markdown", "id": "cell-018", "metadata": {}, "source": [ "The last line shows a trap: SQLite stores no time zone, so an aware timestamp comes back\n", "**naive**. The capstone stores UTC and re-attaches the zone when reading." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 5 }