Advanced20 min

Databases and SQL from Python

Query SQLite from Python, use parameters to avoid injection, and move results into pandas.

Why this matters in AI / ML / GenAI

Training data usually starts in a database. Feature stores, experiment metadata, RAG document stores, and chat history are all tables. Knowing enough SQL from Python to pull, join, and aggregate is a baseline skill for any AI engineer.

sqlite3, and how it maps to everything else

sqlite3 is in the standard library and needs no server, which makes it perfect for learning, tests, and local caches. The API is the Python DB-API 2.0, so psycopg for PostgreSQL and mysql-connector for MySQL work the same way — only the connection string changes.

The flow is always: connect, get a cursor, execute, fetch, commit, close.

Use with sqlite3.connect(path) as conn: so a transaction commits on success and rolls back on an exception.

Fetching: fetchone() for one row, fetchall() for everything, or iterate the cursor for large result sets so you do not load the whole table into memory.

Parameters, not string formatting

Never build SQL with f-strings or +. This is SQL injection, and it is still one of the most exploited vulnerabilities in production systems.

Use placeholders and pass values separately:

cur.execute("SELECT * FROM runs WHERE model = ?", (model_name,))

The driver escapes the value safely. SQLite uses ?; PostgreSQL uses %s; some drivers support named parameters like :model.

This matters doubly in AI applications, where the value often comes from an LLM or a user prompt. An agent that writes SQL from natural language must run against a read-only connection with a restricted user, never with credentials that can drop a table.

executemany inserts many rows in one call and is dramatically faster than looping execute.

Getting data into pandas

pd.read_sql_query(sql, conn) returns a DataFrame directly — the fastest route from a database to analysis or model training.

Push aggregation into the database when the table is large. GROUP BY on the server transfers a handful of rows instead of millions; the same aggregation in pandas requires loading everything first.

For bigger applications, SQLAlchemy provides connection pooling and an ORM. Start with raw SQL; add SQLAlchemy when connection management or model mapping becomes the pain point.

Index the columns you filter and join on. A missing index turns a fast query into a full table scan, and it is the single most common cause of a slow data pipeline.

Copy-paste examples

Copy into your own editor, or load one into the compiler below and press Run.

Create, insert, and query

Runs here — SQLite is loaded on demand into the browser sandbox.

import sqlite3

conn = sqlite3.connect(":memory:")
cur = conn.cursor()

cur.execute("""
    CREATE TABLE runs (
        id INTEGER PRIMARY KEY,
        model TEXT NOT NULL,
        accuracy REAL,
        tokens INTEGER,
        created_at TEXT
    )
""")

rows = [
    ("gpt-4.1-mini", 0.913, 620, "2026-09-01"),
    ("gpt-4.1-mini", 0.907, 540, "2026-09-02"),
    ("llama-3-8b", 0.847, 1520, "2026-09-02"),
    ("llama-3-8b", 0.861, 1480, "2026-09-03"),
    ("claude-haiku", 0.900, 460, "2026-09-03"),
]
cur.executemany(
    "INSERT INTO runs (model, accuracy, tokens, created_at) VALUES (?, ?, ?, ?)", rows
)
conn.commit()

print("rows inserted:", cur.rowcount)
for row in cur.execute("SELECT id, model, accuracy FROM runs ORDER BY accuracy DESC LIMIT 3"):
    print(row)

cur.execute("SELECT COUNT(*), AVG(accuracy) FROM runs")
count, avg = cur.fetchone()
print(f"\n{count} runs, mean accuracy {avg:.4f}")
conn.close()

Parameters prevent SQL injection

The unsafe query deletes the table. Run it and see.

import sqlite3

conn = sqlite3.connect(":memory:")
conn.executescript("""
    CREATE TABLE users (id INTEGER, name TEXT);
    INSERT INTO users VALUES (1, 'priya'), (2, 'arjun');
""")

# Safe: the value is passed separately and escaped by the driver.
malicious = "priya'; DROP TABLE users; --"
safe = conn.execute("SELECT * FROM users WHERE name = ?", (malicious,)).fetchall()
print("safe query result:", safe)
print("table still exists:", conn.execute("SELECT COUNT(*) FROM users").fetchone()[0], "rows")

# Unsafe: string interpolation lets the input become code.
try:
    conn.executescript(f"SELECT * FROM users WHERE name = '{malicious}'")
except sqlite3.Error as err:
    print("error:", err)

remaining = conn.execute(
    "SELECT name FROM sqlite_master WHERE type='table' AND name='users'"
).fetchall()
print("users table after injection:", remaining or "GONE — the table was dropped")
conn.close()

Aggregate in SQL, analyse in pandas

GROUP BY on the server transfers far less data than loading every row.

import pandas as pd
import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE calls (model TEXT, tokens INTEGER, latency_ms INTEGER, ok INTEGER)")
conn.executemany("INSERT INTO calls VALUES (?, ?, ?, ?)", [
    ("mini", 620, 640, 1), ("mini", 540, 580, 1), ("mini", 900, 30000, 0),
    ("large", 1520, 1800, 1), ("large", 1480, 1750, 1),
    ("haiku", 460, 420, 1), ("haiku", 480, 450, 1),
])
conn.commit()

summary = pd.read_sql_query("""
    SELECT model,
           COUNT(*)             AS calls,
           SUM(tokens)          AS total_tokens,
           ROUND(AVG(latency_ms), 1) AS avg_latency,
           ROUND(AVG(ok) * 100, 1)   AS success_pct
    FROM calls
    GROUP BY model
    HAVING COUNT(*) > 1
    ORDER BY total_tokens DESC
""", conn)

print(summary.to_string(index=False))
print("\ntotal tokens across all models:", int(summary["total_tokens"].sum()))
conn.close()

A safe query helper with a row factory

sqlite3.Row lets you access columns by name instead of position.

import sqlite3

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("CREATE TABLE docs (id INTEGER PRIMARY KEY, title TEXT, topic TEXT, score REAL)")
conn.executemany("INSERT INTO docs (title, topic, score) VALUES (?, ?, ?)", [
    ("Intro to MLOps", "mlops", 0.91),
    ("RAG patterns", "genai", 0.87),
    ("Kubernetes basics", "infra", 0.62),
    ("Agent design", "genai", 0.94),
])
conn.commit()

def search_docs(connection, topic, min_score=0.0, limit=10):
    sql = """
        SELECT id, title, topic, score
        FROM docs
        WHERE topic = ? AND score >= ?
        ORDER BY score DESC
        LIMIT ?
    """
    return [dict(row) for row in connection.execute(sql, (topic, min_score, limit))]

for doc in search_docs(conn, "genai", min_score=0.8):
    print(f"{doc['id']}. {doc['title']:20} {doc['topic']:7} {doc['score']:.2f}")

print("\nno matches returns empty list:", search_docs(conn, "unknown"))
conn.close()

Connecting to PostgreSQL (run locally)

Same DB-API flow, different driver and placeholder style.

# pip install "psycopg[binary]" pandas
import os
import pandas as pd
import psycopg

DSN = os.environ["DATABASE_URL"]      # postgresql://user:pass@host:5432/dbname

def fetch_recent_runs(model_name, limit=100):
    sql = """
        SELECT id, model, accuracy, created_at
        FROM runs
        WHERE model = %s
        ORDER BY created_at DESC
        LIMIT %s
    """
    with psycopg.connect(DSN) as conn:
        return pd.read_sql_query(sql, conn, params=(model_name, limit))

# df = fetch_recent_runs("gpt-4.1-mini")
# print(df.head())

Build a small feature table

Try it — in-browser Python

Packages: sqlite3, pandas, numpy

Add a WHERE clause on total_tokens and see the report shrink.

Output

Python runs in your browser. First run downloads the runtime.

Press Run (or Ctrl+Enter) to execute.

CPython in WebAssembly. Stdlib works. NumPy, pandas, scikit-learn and Matplotlib load on demand, and charts render below. No input(), no GPU, no network installs.

Takeaways

  • sqlite3 follows DB-API 2.0, so Postgres and MySQL drivers work the same way.
  • Always use placeholders — never build SQL with f-strings, especially with LLM input.
  • pd.read_sql_query moves results into pandas; aggregate in SQL when tables are large.