SQLite Best Practices for Python

import sqlite3
import os

DB_PATH = os.getenv("DATABASE_PATH", "data.db")

def get_connection() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row           # Dict-like access to rows
    conn.execute("PRAGMA journal_mode=WAL")  # Concurrent reads during writes
    conn.execute("PRAGMA foreign_keys=ON")   # Enforce FK constraints (OFF by default!)
    conn.execute("PRAGMA busy_timeout=5000") # Wait 5s on lock instead of SQLITE_BUSY
    conn.execute("PRAGMA synchronous=NORMAL")# Safe with WAL, better performance
    return conn

def create_order(conn: sqlite3.Connection, customer_name: str, items: list[dict]) -> dict:
    with conn:  # BEGIN ... COMMIT on success, ROLLBACK on exception
        cursor = conn.execute(
            "INSERT INTO orders (customer_name, total_cents, status) VALUES (?, ?, ?)",
            (customer_name, calculate_total(items), "received")
        )
        order_id = cursor.lastrowid

        conn.executemany(
            "INSERT INTO order_items (order_id, menu_item_id, quantity, price_cents) VALUES (?, ?, ?, ?)",
            [(order_id, i["menu_item_id"], i["quantity"], i["price_cents"]) for i in items]
        )

    return get_order_by_id(conn, order_id)

# BAD -- manual commit, no rollback on error, connection leaked
conn = sqlite3.connect("data.db")
conn.execute("INSERT INTO ...")
conn.commit()

def handle_request():
    conn = get_connection()
    try:
        with conn:
            # ... write operations ...
        return result
    finally:
        conn.close()

# GOOD -- parameterized with ? placeholders
conn.execute("SELECT * FROM orders WHERE id = ?", (order_id,))
conn.execute("SELECT * FROM items WHERE name LIKE ?", (f"%{search}%",))

# BAD -- SQL injection vulnerability
conn.execute(f"SELECT * FROM orders WHERE id = {order_id}")
conn.execute("SELECT * FROM items WHERE name LIKE '%{}%'".format(search))

conn.row_factory = sqlite3.Row
row = conn.execute("SELECT * FROM orders WHERE id = ?", (1,)).fetchone()
row["customer_name"]  # Access by column name
dict(row)             # Convert to plain dict

def get_all_items(conn: sqlite3.Connection) -> list[dict]:
    rows = conn.execute("SELECT * FROM items ORDER BY name").fetchall()
    return [dict(row) for row in rows]

# GOOD -- single call, efficient
conn.executemany(
    "INSERT INTO order_items (order_id, product_id, quantity) VALUES (?, ?, ?)",
    [(order_id, item["id"], item["qty"]) for item in items]
)

# BAD -- N round-trips
for item in items:
    conn.execute("INSERT INTO order_items ...")

# Store in cents to avoid floating-point errors
total_cents = int(price_dollars * 100)
# Display
display_price = f"${total_cents / 100:.2f}"

price_cents INTEGER NOT NULL  -- NOT REAL

created_at TEXT DEFAULT (datetime('now')) NOT NULL

from datetime import datetime
created = datetime.fromisoformat(row["created_at"])

CREATE INDEX idx_order_items_order_id ON order_items(order_id);
CREATE INDEX idx_movements_product_id ON movements(product_id);

status TEXT NOT NULL DEFAULT 'received'
    CHECK (status IN ('received', 'preparing', 'ready', 'cancelled'))

CREATE TABLE order_items (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    order_id INTEGER NOT NULL REFERENCES orders(id) ON DELETE CASCADE,
    -- ...
);
-- Remember: CASCADE only works when PRAGMA foreign_keys=ON

# Option 1: New connection per request (recommended)
def get_db():
    conn = get_connection()
    try:
        yield conn
    finally:
        conn.close()

# Option 2: Allow cross-thread (less safe)
conn = sqlite3.connect(DB_PATH, check_same_thread=False)

def init_db():
    conn = get_connection()
    try:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS _migrations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL UNIQUE,
                applied_at TEXT DEFAULT (datetime('now'))
            )
        """)

        migrations = [
            ("001_initial", """
                CREATE TABLE products (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL,
                    price_cents INTEGER NOT NULL,
                    created_at TEXT DEFAULT (datetime('now')) NOT NULL
                );
                CREATE INDEX idx_products_name ON products(name);
            """),
        ]

        for name, sql in migrations:
            if not conn.execute("SELECT 1 FROM _migrations WHERE name = ?", (name,)).fetchone():
                conn.executescript(sql)
                conn.execute("INSERT INTO _migrations (name) VALUES (?)", (name,))
                conn.commit()
    finally:
        conn.close()