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.
1. Read
Understand the idea in plain English first.
2. Run
Load any example into the compiler and press Run.
3. Change
Edit one value, rerun, and learn from the output.
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.
Hands-on practice
Compiler on the left. Examples on the right.
On desktop, keep the compiler beside the examples. On mobile, the same blocks stack cleanly. Pick an example, try it in the compiler, then change one small thing.
Build a small feature table
Live Python compiler
Try it — in-browser Python
Loads when needed: sqlite3, pandas, numpy
Add a WHERE clause on total_tokens and see the report shrink.
Code editor
Output
Python runs in your browser. First run downloads the runtime.
Press Run (or Ctrl+Enter) to execute.
Runs CPython in your browser. NumPy, pandas, scikit-learn and Matplotlib load on demand. Charts appear below the output. No input(), no GPU, no network installs.
Clear code examples
Every example is copy-ready. Use Try in compiler when you want to experiment without scrolling around.
Example
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()Example
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()Example
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()Example
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()Example
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())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.