Database Programming Proficient¶
When you'd use this
sqlite3, PostgreSQL, connection pooling, transactions, migrations and patterns.
Talk to databases from Python — connections, parameterized queries, transactions — for any app that persists data.
sqlite3 (built-in) — complete guide¶
A zero-setup file database bundled with Python — perfect for local apps, tests, and prototypes.
CRUD operations¶
Create, read, update, and delete rows — the four operations behind almost every data-backed app. Note every query uses ? placeholders, never string formatting, to stay injection-safe.
import sqlite3
# Connect (creates file if not exists)
conn = sqlite3.connect("app.db")
conn.row_factory = sqlite3.Row # access columns by name
cursor = conn.cursor()
# ─── CREATE TABLE ─────────────────────────────────
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
age INTEGER CHECK(age >= 0),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# ─── INSERT (parameterized — prevents SQL injection!) ─
cursor.execute(
"INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
("Alice", "alice@example.com", 30)
)
# Insert many rows
users = [
("Bob", "bob@example.com", 25),
("Charlie", "charlie@example.com", 35),
("Diana", "diana@example.com", 28),
]
cursor.executemany(
"INSERT INTO users (name, email, age) VALUES (?, ?, ?)", users
)
# ─── SELECT ───────────────────────────────────────
cursor.execute("SELECT * FROM users WHERE age > ?", (26,))
for row in cursor.fetchall():
print(f"{row['name']} ({row['email']}) - age {row['age']}")
# Output:
# Alice (alice@example.com) - age 30
# Charlie (charlie@example.com) - age 35
# Diana (diana@example.com) - age 28
# Single row
cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
user = cursor.fetchone()
print(dict(user)) # {'id': 1, 'name': 'Alice', ...}
# ─── UPDATE ───────────────────────────────────────
cursor.execute(
"UPDATE users SET age = ? WHERE name = ?", (31, "Alice")
)
print(f"Rows affected: {cursor.rowcount}") # 1
# ─── DELETE ───────────────────────────────────────
cursor.execute("DELETE FROM users WHERE name = ?", ("Bob",))
conn.commit()
conn.close()
Context manager pattern¶
Wrap connection setup, commit/rollback, and close in a with block so every caller gets automatic cleanup — commit on success, rollback on exception, close no matter what.
from contextlib import contextmanager
@contextmanager
def get_db(path="app.db"):
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL") # better concurrency
conn.execute("PRAGMA foreign_keys=ON") # enforce FK constraints
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
# Usage
with get_db() as db:
db.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
("Eve", "eve@example.com", 22))
# Auto-commits on success, auto-rolls back on exception
Transactions¶
Group multiple writes so they either all succeed or all roll back — essential for operations like money transfers where a partial update would corrupt your data.
with get_db() as db:
try:
db.execute("BEGIN")
db.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
db.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
db.execute("COMMIT")
except Exception:
db.execute("ROLLBACK")
raise
PostgreSQL with psycopg (v3)¶
Connect to production-grade PostgreSQL — connections, parameterized queries, and transactions.
import psycopg
from psycopg.rows import dict_row
# ─── Connection ───────────────────────────────────
conn = psycopg.connect(
"postgresql://user:pass@localhost:5432/mydb",
row_factory=dict_row,
)
# ─── Queries with named parameters ───────────────
with conn.cursor() as cur:
cur.execute(
"SELECT * FROM users WHERE age > %(min_age)s AND city = %(city)s",
{"min_age": 25, "city": "NYC"},
)
users = cur.fetchall()
for user in users:
print(user["name"], user["email"])
conn.commit()
conn.close()
Connection pooling¶
Reuse a fixed set of open connections instead of opening a new one per request — opening a Postgres connection is expensive, so pooling is critical for throughput under load.
from psycopg_pool import ConnectionPool
pool = ConnectionPool(
"postgresql://user:pass@localhost/mydb",
min_size=5,
max_size=20,
)
def get_users():
with pool.connection() as conn:
with conn.cursor(row_factory=dict_row) as cur:
cur.execute("SELECT * FROM users")
return cur.fetchall()
# Pool handles connection reuse, health checks, etc.
Async PostgreSQL¶
Query without blocking the event loop — use the async API inside async def handlers (FastAPI, aiohttp) so one worker can serve many concurrent requests while queries are in flight.
import asyncio
import psycopg
from psycopg.rows import dict_row
async def get_user(user_id: int):
async with await psycopg.AsyncConnection.connect(
"postgresql://user:pass@localhost/mydb",
row_factory=dict_row,
) as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
return await cur.fetchone()
Query patterns¶
Common, safe query shapes — always parameterize to prevent SQL injection.
Parameterized queries (ALWAYS use these)¶
Pass user input as parameters, never by string-formatting it into the SQL — the driver escapes values safely, which is the single most important defense against SQL injection.
# SAFE — parameterized
cursor.execute("SELECT * FROM users WHERE name = ?", (user_input,))
# DANGEROUS — string formatting (SQL injection!)
cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'") # NEVER!
Bulk operations¶
Insert thousands of rows in one call with executemany instead of looping — far fewer round-trips to the database, so batch loads run in a fraction of the time.
# executemany — efficient batch insert
data = [(f"user_{i}", f"user{i}@example.com", 20 + i) for i in range(10000)]
with get_db() as db:
db.executemany(
"INSERT INTO users (name, email, age) VALUES (?, ?, ?)", data
)
# Much faster than 10000 individual inserts
Full-text search (SQLite FTS5)¶
Search text columns by relevance (not just exact matches) using SQLite's built-in FTS5 engine — add keyword search to an app without standing up Elasticsearch.
with get_db() as db:
db.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS articles_fts
USING fts5(title, content)
""")
db.execute(
"INSERT INTO articles_fts (title, content) VALUES (?, ?)",
("Python Tips", "Learn decorators and generators...")
)
# Search
results = db.execute(
"SELECT * FROM articles_fts WHERE articles_fts MATCH ?",
("decorators",)
).fetchall()
Schema migrations¶
Evolve the database schema over time in versioned, repeatable steps.
Manual approach¶
Track a schema version number in the database and apply only the migrations newer than it — a dependency-free way to evolve your schema repeatably across environments.
MIGRATIONS = [
"""CREATE TABLE IF NOT EXISTS schema_version (version INTEGER)""",
"""INSERT INTO schema_version VALUES (0)""",
"""ALTER TABLE users ADD COLUMN phone TEXT""",
"""CREATE INDEX idx_users_email ON users(email)""",
]
def migrate(db):
db.execute("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER DEFAULT 0)")
row = db.execute("SELECT MAX(version) as v FROM schema_version").fetchone()
current = row["v"] or 0
for i, sql in enumerate(MIGRATIONS[current:], start=current):
print(f" Applying migration {i + 1}...")
db.execute(sql)
db.execute("UPDATE schema_version SET version = ?", (i + 1,))
db.commit()
Practice Exercises¶
- Build a complete CRUD module for a
taskstable with proper error handling and transactions. - Implement connection pooling for a multi-threaded application.
- Write a migration system that applies
.sqlfiles in order. - Build a full-text search feature for a blog using SQLite FTS5.
- Compare performance of individual inserts vs
executemanyvs COPY (PostgreSQL) for 100K rows. - Implement soft-delete (set
deleted_attimestamp instead of actually deleting).
💬 Discussion
Have a question about this topic? Found an error? Share your thoughts below.