Chapter 18 · Capstone: Design and Query a Complete Database
Add Transactions, Indexes, Security, and Backup Procedures
Correct tables and queries are not a production system. This lesson adds failure boundaries, concurrency-aware inventory reservation, duplicate-request protection, query-plan evidence, access controls, parameter binding, backup procedures, and restore tests.
Learning outcomes
Learning outcomes
Implement atomic, idempotent order placement with inventory reservation and explicit failure behavior.
Choose indexes from query predicates, joins, ordering, uniqueness, and measured plans.
Separate read, write, administrative, and analytical privileges in a production design.
Use parameter binding and allow-listed identifiers at the application boundary.
Create, restore, verify, and document backups against recovery objectives.
Production hardening model
Transaction boundary
All state changes that make one business action true commit together or roll back together.
Access path
Indexes support measured read patterns while adding write, storage, maintenance, and deployment cost.
Trust boundary
Roles, views, parameter binding, secrets, and audit evidence reduce what each actor can do.
Recovery system
Backups matter only when restore procedures are tested against defined loss and downtime targets.
The request either reaches one durable state or leaves no partial business action.
Atomic and idempotent order placement
import sqlite3from dataclasses import dataclass@dataclass(frozen=True)class Line: product_id: int quantity: intdef place_order(db: sqlite3.Connection, *, request_key: str, order_number: str, customer_id: int, address_id: int, warehouse_id: int, lines: list[Line]) -> int: db.execute('PRAGMA foreign_keys = ON') db.execute('BEGIN IMMEDIATE') try: existing = db.execute( 'SELECT order_id FROM sales_order WHERE request_key = ?', (request_key,), ).fetchone() if existing: db.commit() return existing[0] address_owner = db.execute( 'SELECT customer_id FROM address WHERE address_id = ?', (address_id,), ).fetchone() if not address_owner or address_owner[0] != customer_id: raise ValueError('shipping address does not belong to customer') if not lines: raise ValueError('order requires at least one line') order_id = db.execute( '''INSERT INTO sales_order( order_number,customer_id,shipping_address_id, status,currency_code,ordered_at,request_key ) VALUES(?,?,?,'submitted','USD',CURRENT_TIMESTAMP,?) RETURNING order_id''', (order_number, customer_id, address_id, request_key), ).fetchone()[0] for line_no, line in enumerate(lines, 1): product = db.execute( 'SELECT unit_price_cents, active FROM product WHERE product_id = ?', (line.product_id,), ).fetchone() if not product or product[1] != 1: raise ValueError(f'product unavailable: {line.product_id}') changed = db.execute( '''UPDATE inventory SET reserved = reserved + ?, updated_at = CURRENT_TIMESTAMP WHERE warehouse_id = ? AND product_id = ? AND on_hand - reserved >= ?''', (line.quantity, warehouse_id, line.product_id, line.quantity), ).rowcount if changed != 1: raise ValueError(f'insufficient stock: {line.product_id}') db.execute( '''INSERT INTO order_item( order_id,line_no,product_id,quantity,unit_price_cents ) VALUES(?,?,?,?,?)''', (order_id, line_no, line.product_id, line.quantity, product[0]), ) db.execute( '''INSERT INTO audit_event( entity_type,entity_id,action,actor,occurred_at,details ) VALUES('sales_order',?,'created','api:checkout',CURRENT_TIMESTAMP,?)''', (str(order_id), f'{{"request_key":"{request_key}"}}'), ) db.commit() return order_id except Exception: db.rollback() raise| Failure | Expected outcome |
|---|---|
| Repeated request_key | Return the original order; do not reserve again |
| Insufficient stock on any line | No order, line, audit event, or reservation remains |
| Address owned by another customer | Reject before durable changes |
| Concurrent reservation | BEGIN IMMEDIATE serializes writers; predicate rechecks available stock |
| Unexpected exception | Rollback the complete unit of work and surface an observable error |
Transaction tests
def test_order_transaction(db): before = db.execute( 'SELECT reserved FROM inventory WHERE warehouse_id=2 AND product_id=12' ).fetchone()[0] first = place_order( db, request_key='req-capstone-1', order_number='NS-2026-CAP-1', customer_id=2, address_id=201, warehouse_id=2, lines=[Line(12, 2)], ) second = place_order( db, request_key='req-capstone-1', order_number='ignored-on-retry', customer_id=2, address_id=201, warehouse_id=2, lines=[Line(12, 2)], ) assert first == second after = db.execute( 'SELECT reserved FROM inventory WHERE warehouse_id=2 AND product_id=12' ).fetchone()[0] assert after == before + 2 try: place_order( db, request_key='req-capstone-fail', order_number='NS-2026-CAP-X', customer_id=2, address_id=201, warehouse_id=2, lines=[Line(12, 999)], ) except ValueError: pass assert db.execute( "SELECT COUNT(*) FROM sales_order WHERE request_key='req-capstone-fail'" ).fetchone()[0] == 0Index workload from evidence
| Workload | Predicate/order shape | Candidate index |
|---|---|---|
| Customer lookup | email = ? | UNIQUE customer(email) already supplies it |
| Customer order history | customer_id = ? ORDER BY ordered_at DESC, order_id DESC | (customer_id, ordered_at DESC, order_id DESC) |
| Global recent-order pagination | ORDER BY ordered_at DESC, order_id DESC | (ordered_at DESC, order_id DESC) |
| Fulfillment queue | status IN (...) ORDER BY ordered_at, order_id | partial or composite status/order index |
| Product sales joins | order_item.product_id = ? | (product_id, order_id) |
| Payment reconciliation | payment.order_id = ? | (order_id, status) |
| Shipment lookup | shipment.order_id = ? | (order_id, status) |
| Audit investigation | entity_type + entity_id ORDER BY occurred_at | (entity_type, entity_id, occurred_at DESC) |
CREATE INDEX idx_order_customer_history ON sales_order(customer_id, ordered_at DESC, order_id DESC);CREATE INDEX idx_order_recent ON sales_order(ordered_at DESC, order_id DESC);CREATE INDEX idx_order_fulfillment ON sales_order(status, ordered_at, order_id) WHERE status IN ('submitted','paid','packed');CREATE INDEX idx_order_item_product ON order_item(product_id, order_id);CREATE INDEX idx_payment_order_status ON payment(order_id, status);CREATE INDEX idx_shipment_order_status ON shipment(order_id, status);CREATE INDEX idx_audit_entity_time ON audit_event(entity_type, entity_id, occurred_at DESC);Read the plans before and after
EXPLAIN QUERY PLANSELECT order_id, order_number, status, ordered_atFROM sales_orderWHERE customer_id = 1ORDER BY ordered_at DESC, order_id DESCLIMIT 20;EXPLAIN QUERY PLANSELECT order_id, order_number, status, ordered_atFROM sales_orderWHERE status IN ('submitted','paid','packed')ORDER BY ordered_at, order_id;Record the query, representative parameters, row counts, database version, plan before, plan after, and measured timing. An index is accepted because it improves an important workload at tolerable write cost—not because its column names look plausible.
Least-privilege production design
CREATE ROLE northstar_readonly NOLOGIN;CREATE ROLE northstar_app NOLOGIN;CREATE ROLE northstar_analyst NOLOGIN;GRANT USAGE ON SCHEMA northstar TO northstar_readonly, northstar_app, northstar_analyst;GRANT SELECT ON ALL TABLES IN SCHEMA northstar TO northstar_readonly;GRANT SELECT, INSERT, UPDATE ON northstar.sales_order, northstar.order_item, northstar.payment, northstar.shipment, northstar.audit_eventTO northstar_app;GRANT SELECT ON northstar.reporting_monthly_revenue TO northstar_analyst;REVOKE DELETE ON northstar.audit_event FROM northstar_app;REVOKE ALL ON SCHEMA public FROM PUBLIC;| Principal | Allowed | Denied by design |
|---|---|---|
| Application runtime | Bound operational reads/writes needed by services | DDL, role management, unrestricted deletes, backup administration |
| Customer service | Approved views and procedures | Raw payment/audit tables and bulk exports |
| Analyst | Curated reporting views | Operational mutation and sensitive raw attributes |
| Migration role | Versioned DDL during controlled deployment | Normal application login |
| Backup operator | Backup/restore capabilities and encrypted storage access | Application business operations |
Parameterized access boundary
ALLOWED_SORTS = { 'newest': 'ordered_at DESC, order_id DESC', 'oldest': 'ordered_at ASC, order_id ASC', 'number': 'order_number ASC',}def list_orders(db, customer_id: int, sort_name: str, limit: int = 50): order_by = ALLOWED_SORTS.get(sort_name) if order_by is None: raise ValueError('unsupported sort') limit = max(1, min(limit, 100)) sql = f'''SELECT order_id, order_number, status, ordered_at FROM sales_order WHERE customer_id = ? ORDER BY {order_by} LIMIT ?''' return db.execute(sql, (customer_id, limit)).fetchall()Bind data values. SQL identifiers and syntax positions generally cannot be bound, so select them from a fixed allow-list rather than concatenating user input.
Backup and restore procedure
import sqlite3from pathlib import Pathdef create_verified_backup(source_path: str, backup_path: str) -> None: Path(backup_path).unlink(missing_ok=True) with sqlite3.connect(source_path) as source, sqlite3.connect(backup_path) as target: source.backup(target) with sqlite3.connect(backup_path) as restored: restored.execute('PRAGMA foreign_keys = ON') assert restored.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' assert restored.execute('PRAGMA foreign_key_check').fetchall() == [] expected = {'customer': 4, 'product': 6, 'sales_order': 6} for table_name, count in expected.items(): actual = restored.execute(f'SELECT COUNT(*) FROM {table_name}').fetchone()[0] assert actual == count, (table_name, actual, count)create_verified_backup('northstar.db', 'backups/northstar-2026-08-05.db')set -euo pipefailstamp=$(date -u +%Y%m%dT%H%M%SZ)pg_dump --format=custom --no-owner --file="northstar-${stamp}.dump" northstarcreatedb northstar_restore_testpg_restore --exit-on-error --clean --if-exists --dbname=northstar_restore_test "northstar-${stamp}.dump"psql northstar_restore_test -v ON_ERROR_STOP=1 -f db/queries/verification.sqlRecovery objectives and runbook evidence
| Runbook field | Capstone value |
|---|---|
| Recovery point objective | Maximum accepted data loss: 24 hours for the local capstone; production target must reflect business value |
| Recovery time objective | Restore and verification completed within 30 minutes in the exercise environment |
| Backup schedule | Daily logical/copy backup plus pre-migration backup |
| Retention | Seven daily and four weekly copies for the exercise; encryption and off-host storage required in production |
| Restore target | Always restore into a separate path/database first |
| Verification | Integrity check, foreign-key check, row counts, reconciliation queries, application smoke test |
| Ownership | Named backup operator; incident commander authorizes production cutover |
Hardening review
- Why is request_key part of the transaction design?
- Why is reserved <= on_hand insufficient by itself for concurrent order placement?
- When should an index proposal be rejected?
- Why is creating a backup not proof of recoverability?
Review the answers
The request key makes retries map to one durable business action.
A transaction and conditional update are needed to serialize and re-check availability.
Reject indexes that do not improve an important measured workload or whose write/storage cost is unjustified.
Recoverability requires a separate restore plus integrity and business verification.
Lesson summary
- Transactions define one durable business action and its all-or-nothing failure boundary.
- Idempotency prevents retries from duplicating orders and reservations.
- Indexes are workload evidence with ongoing costs, not decorations.
- Least privilege separates runtime, analyst, migration, and backup capabilities.
- Backups become reliable only through repeated restore verification and a named runbook.