04 — db.py: SQLite + safe_query

Duration: 18 min Prerequisites: chapter 03 (you have a normalized DataFrame), basic SQL.

What db.py owns

db.py is the only module in csv-llm-shared that talks to SQLite. It exposes 3 public functions:

db.open_db(path)             → sqlite3.Connection
db.load_dataframe(conn, df, profile)  → int (rows inserted)
db.safe_query(conn, sql)     → pd.DataFrame  (SELECT-only)

Apps import db and use those three. Nothing else.

The schema

csv-llm-shared/db.py

import re
import sqlite3
from pathlib import Path
import pandas as pd

SCHEMA = """
CREATE TABLE IF NOT EXISTS transactions (
    profile     TEXT NOT NULL,
    date        TEXT NOT NULL,
    card        TEXT,
    description TEXT NOT NULL,
    category    TEXT NOT NULL,
    amount      REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_profile_date ON transactions(profile, date);
CREATE INDEX IF NOT EXISTS idx_category    ON transactions(category);
"""

Six columns. Two indices:

• (profile, date) — covers every query that filters by profile and groups/sorts by date.
• (category) — covers GROUP BY category queries.

That’s it. No id, no foreign keys, no triggers. Deliberately minimal.

Why no PRIMARY KEY?

SQLite implicitly creates a hidden rowid for every row. We don’t need a surface-level primary key because we never UPDATE a single row by id — we only INSERT after DELETE WHERE profile = '...'. The hidden rowid is enough.

open_db — create or open

def open_db(path: str) -> sqlite3.Connection:
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(path)
    conn.executescript(SCHEMA)
    return conn

Three steps:

1. Make sure the parent folder exists (e.g. .cache/).
2. Open a SQLite connection. SQLite creates the file if it doesn’t exist.
3. Run the schema (idempotent thanks to IF NOT EXISTS).

We don’t open in read-only mode here, because we need to be able to INSERT. The safe_query guard handles the read-only enforcement instead.

load_dataframe — DELETE then INSERT

def load_dataframe(conn, df: pd.DataFrame, profile: str) -> int:
    """(Re)load a profile: delete existing rows then insert."""
    conn.execute("DELETE FROM transactions WHERE profile = ?", (profile,))
    df.to_sql("transactions", conn, if_exists="append", index=False)
    conn.commit()
    return len(df)

This is the dedup strategy: instead of trying to figure out which rows are new vs existing, we wipe the profile clean and re-insert everything. Pros:

• Trivially correct.
• Idempotent: same CSV in → same DB out, every time.
• Fast for our scale (734 rows × 6 columns = a few ms).

Cons:

• Doesn’t scale to millions of rows per profile (would need an incremental strategy).
• Doesn’t preserve any user-edited categories (we don’t have user edits, so fine).

safe_query — the SELECT-only guard

FORBIDDEN = re.compile(
    r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|REPLACE|ATTACH|DETACH|PRAGMA)\b",
    re.IGNORECASE,
)


def safe_query(conn, sql: str) -> pd.DataFrame:
    """SELECT-only. Raises ValueError on anything else."""
    s = sql.strip()
    if not s.lower().startswith("select") and not s.lower().startswith("with"):
        raise ValueError("SQL refused: only SELECT statements allowed.")
    if FORBIDDEN.search(s):
        raise ValueError("SQL refused: forbidden keyword detected.")
    # Detect multi-statement (single trailing `;` is OK)
    if ";" in s.rstrip().rstrip(";"):
        raise ValueError("SQL refused: multiple statements not allowed.")
    return pd.read_sql_query(s, conn)

This is the function the LLM gets to call as a tool. We must be sure it can’t be tricked into mutating data.

The three defenses

1. Must start with SELECT or WITH. Anything else → reject. The WITH is for common table expressions: WITH recent AS (SELECT ...) SELECT * FROM recent.

2. No forbidden keyword anywhere. Even inside a string literal (we don’t bother to parse SQL properly — better to over-reject than miss a bypass).

3. No multi-statement. A clever LLM might produce SELECT 1; DROP TABLE transactions;. The first ; between statements is not in the trailing position, so we reject.

Why not just open in read-only mode?

You can: sqlite3.connect("file:db?mode=ro", uri=True). But:

• Then you can’t load_dataframe() from the same connection (you’d need two connections).
• The safe_query guard also rejects PRAGMA, ATTACH, etc. that are read-ish but can leak data (PRAGMA database_list).

So we use guard + (optionally) read-only mode for defense in depth. In production, do both.

Attacks safe_query blocks

Try these — they all ValueError:

>>> safe_query(conn, "INSERT INTO transactions VALUES (...)")
ValueError: SQL refused: only SELECT statements allowed.

>>> safe_query(conn, "SELECT 1; DROP TABLE transactions")
ValueError: SQL refused: multiple statements not allowed.

>>> safe_query(conn, "SELECT * FROM transactions; -- DROP TABLE x")
# Tricky: -- is a SQL comment. After comment stripping, this is one statement.
# Our regex catches DROP regardless.
ValueError: SQL refused: forbidden keyword detected.

>>> safe_query(conn, "/* nothing */ DELETE FROM transactions")
ValueError: SQL refused: only SELECT statements allowed.

Attacks that get through (and why we accept them)

>>> safe_query(conn, "SELECT * FROM sqlite_master")
# Returns the schema. We're OK with this — it's just listing tables.

>>> safe_query(conn, "SELECT * FROM transactions WHERE category='Housing'")
# Returns 84 rows of legitimate data. Intended.

If you want to also prevent schema enumeration, add a fourth defense: an explicit whitelist of tables the query may reference. We don’t, because for a personal-spending app, schema leaks aren’t a real threat.

Live test

>>> import sys; sys.path.insert(0, "csv-llm-shared")
>>> import ingest, normalize, db
>>> df = ingest.read_csv("csv-llm-shared/data/data1-anonymized.csv")
>>> nf = normalize.normalize(df, profile="data1")
>>> conn = db.open_db("/tmp/test.sqlite")
>>> db.load_dataframe(conn, nf, profile="data1")
734
>>> db.safe_query(conn, """
...   SELECT category, SUM(amount) AS total
...   FROM transactions
...   WHERE profile = 'data1' AND amount > 0
...   GROUP BY category
...   ORDER BY total DESC
...   LIMIT 5
... """)
       category     total
0       Housing  30895.29
1        Travel  16597.44
2      Business  14386.39
3     Groceries   9725.93
4     Education   8987.85

The same top-5 we predicted in chapter 02. Pipeline confirmed.

A real-world misuse and how we’d handle it

Scenario: a teammate writes a one-off script and runs safe_query(conn, "UPDATE transactions SET category='Other' WHERE category=''"). Our guard rejects it. They get annoyed and bypass by writing conn.execute(...) directly.

Three responses:

1. Add a load_dataframe-like helper for legit mutations: db.recategorize(conn, where_sql, new_category). Provides a sanctioned path.
2. Document the policy: “any mutation must go through a named helper, never raw SQL”. Codify in a README.
3. Code review: the diff that adds a raw conn.execute("UPDATE ...") should be caught at PR time.

safe_query protects the LLM from mutating data. It can’t protect against malicious human developers (nothing can).

Performance notes

For 734 rows, every query is sub-millisecond. For 100,000 rows (a year of heavy use across many cards):

• idx_profile_date makes profile-scoped queries instant.
• idx_category makes GROUP BY category instant.
• A full-table scan would take ~10 ms on a modern laptop.

If you ever reach a million rows, two things to add:

1. An index on (description) for top-merchant queries.
2. A partial index WHERE amount > 0 to skip refunds in expense queries.

Takeaways

• 6-column table, 2 indices, no fancy schema.
• load_dataframe uses DELETE-then-INSERT per profile — trivially correct.
• safe_query has three defenses: SELECT-only, no forbidden keywords, no multi-statement.
• The guard makes the database safe to expose to LLM-generated SQL.
• In production, layer read-only DB mode on top for defense in depth.

Next: categorize.py — rules + LLM →