Working with Databases in Python: SQLite, PostgreSQL and SQLAlchemy
Python provides relational database access through the DB-API 2.0 standard (PEP 249) — a common interface across all adapters. SQLAlchemy adds a higher-level abstraction for portable SQL expressions or a full ORM (covered in the SQLAlchemy ORM guide).
sqlite3 — built-in database
import sqlite3
conn = sqlite3.connect('app.db')
# Improve concurrency and enforce foreign keys
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
# Context manager — auto-commit on exit, rollback on exception
with conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS files (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
format TEXT NOT NULL,
size INTEGER,
created TEXT DEFAULT (datetime('now'))
)
""")
# Parameterized INSERT (prevents SQL injection)
with conn:
conn.execute("INSERT INTO files (name, format, size) VALUES (?, ?, ?)",
('photo.jpg', 'JPEG', 2_456_789))
# Bulk insert efficiently
rows = [('video.mp4', 'MP4', 104_857_600), ('doc.pdf', 'PDF', 512_000)]
conn.executemany("INSERT OR IGNORE INTO files (name, format, size) VALUES (?, ?, ?)", rows)
# Query
conn.row_factory = sqlite3.Row # dict-like access
cur = conn.execute("SELECT * FROM files ORDER BY name")
files = [dict(r) for r in cur]
conn.close()
Context manager helper
from contextlib import contextmanager
import sqlite3
@contextmanager
def get_db(path: str = 'app.db'):
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
with get_db() as db:
db.execute("INSERT INTO files (name, format) VALUES (?, ?)", ('test.png', 'PNG'))
psycopg2 — PostgreSQL
pip install psycopg2-binary
import psycopg2
import psycopg2.extras
DSN = "host=localhost port=5432 dbname=mydb user=postgres password=secret"
conn = psycopg2.connect(DSN)
conn.autocommit = False
try:
with conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS files (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL,
format TEXT NOT NULL,
size BIGINT,
processed BOOLEAN DEFAULT FALSE,
created_at TIMESTAMPTZ DEFAULT NOW()
)
""")
conn.commit()
# INSERT with RETURNING to get the new id
cur.execute(
"INSERT INTO files (name, format, size) VALUES (%s, %s, %s) RETURNING id",
('photo.jpg', 'JPEG', 2_456_789)
)
new_id = cur.fetchone()[0]
print(f"Inserted file id={new_id}")
# Bulk insert with execute_values (much faster than executemany)
data = [('video.mp4', 'MP4', 104_857_600), ('doc.pdf', 'PDF', 512_000)]
psycopg2.extras.execute_values(
cur,
"INSERT INTO files (name, format, size) VALUES %s",
data,
page_size=100,
)
conn.commit()
# Query with RealDictCursor
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as dcur:
dcur.execute("SELECT * FROM files WHERE format=%s ORDER BY created_at DESC", ('JPEG',))
for row in dcur:
print(dict(row))
except psycopg2.Error as e:
conn.rollback()
print(f"PostgreSQL error: {e.pgcode} — {e.pgerror}")
raise
finally:
conn.close()
Connection pooling
from psycopg2 import pool
connection_pool = pool.ThreadedConnectionPool(minconn=2, maxconn=10, dsn=DSN)
def query(sql: str, params: tuple = ()) -> list[dict]:
conn = connection_pool.getconn()
try:
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(sql, params)
if cur.description:
return [dict(r) for r in cur.fetchall()]
conn.commit()
return []
except Exception:
conn.rollback()
raise
finally:
connection_pool.putconn(conn)
rows = query("SELECT * FROM files WHERE format=%s", ('JPEG',))
SQLAlchemy Core — portable, expressive SQL
pip install sqlalchemy
from sqlalchemy import (
create_engine, MetaData, Table, Column,
Integer, String, BigInteger, Boolean, DateTime,
select, insert, update, delete, and_, func,
)
from datetime import datetime
engine = create_engine(
"sqlite:///app.db",
# "postgresql+psycopg2://user:pass@localhost/db",
echo=False,
pool_size=5,
max_overflow=10,
pool_pre_ping=True, # re-validates connections after idle time
)
metadata = MetaData()
files = Table('files', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(255),nullable=False),
Column('format', String(10), nullable=False),
Column('size', BigInteger),
Column('processed', Boolean, default=False),
Column('created_at',DateTime, default=datetime.now),
)
metadata.create_all(engine)
with engine.connect() as conn:
# INSERT single row
conn.execute(insert(files).values(name='image.png', format='PNG', size=1_234_567))
# INSERT multiple rows
conn.execute(insert(files), [
{'name': 'audio.mp3', 'format': 'MP3', 'size': 5_000_000},
{'name': 'video.webm', 'format': 'WEBM', 'size': 25_000_000},
])
conn.commit()
# SELECT with filters
stmt = (
select(files)
.where(and_(
files.c.format.in_(['PNG', 'JPEG']),
files.c.size > 100_000,
))
.order_by(files.c.size.desc())
.limit(10)
)
for row in conn.execute(stmt).mappings():
print(dict(row))
# UPDATE
conn.execute(update(files).where(files.c.format == 'PNG').values(processed=True))
conn.commit()
# Aggregations
agg = select(
files.c.format,
func.count().label('total'),
func.sum(files.c.size).label('total_bytes'),
).group_by(files.c.format).order_by(func.count().desc())
for row in conn.execute(agg):
print(f"{row.format}: {row.total} files, {row.total_bytes/1024:.1f} KB")
Transactions and savepoints
from sqlalchemy.exc import IntegrityError
with engine.begin() as conn: # auto commit/rollback
conn.execute(insert(files).values(name='a.jpg', format='JPEG'))
# Savepoint — partial rollback without undoing the whole transaction
savepoint = conn.begin_nested()
try:
conn.execute(insert(files).values(name='a.jpg', format='JPEG')) # duplicate
savepoint.commit()
except IntegrityError:
savepoint.rollback()
print("Duplicate ignored — transaction continues")
conn.execute(insert(files).values(name='b.jpg', format='JPEG'))
# commit happens automatically at end of `with engine.begin()`
Schema migrations with Alembic
pip install alembic
alembic init migrations
# migrations/env.py — point to your metadata
from myapp.models import metadata
target_metadata = metadata
# Auto-generate a migration (detects schema changes)
# alembic revision --autogenerate -m "add_comment_column"
# Example generated migration
from alembic import op
import sqlalchemy as sa
def upgrade():
op.add_column('files', sa.Column('comment', sa.Text(), nullable=True))
op.create_index('ix_files_format', 'files', ['format'])
def downgrade():
op.drop_index('ix_files_format', table_name='files')
op.drop_column('files', 'comment')
# Apply / revert migrations
# alembic upgrade head
# alembic downgrade -1
# alembic history --verbose
SQL injection prevention
# ✅ CORRECT — always use parameterized queries
name = input("Username: ")
# sqlite3
cur.execute("SELECT * FROM users WHERE name=?", (name,))
# psycopg2
cur.execute("SELECT * FROM users WHERE name=%s", (name,))
# SQLAlchemy Core (any database)
conn.execute(select(users).where(users.c.name == name))
# ❌ NEVER format SQL manually — vulnerable to injection
# cur.execute(f"SELECT * FROM users WHERE name='{name}'")
Best practices
- Always use parameterized queries (?, %s, or SQLAlchemy expressions) — never concatenate variables into raw SQL strings.
with engine.begin()in SQLAlchemy for automatic commit/rollback on the whole block.- Connection pooling is essential in web apps — avoids opening a new connection per request (expensive and slow).
- Enable
PRAGMA foreign_keys=ONin SQLite — it's OFF by default, so foreign key constraints are silently ignored without it. - Use Alembic for migrations — never alter the production schema manually; use versioned, reviewable migration scripts.
pool_pre_ping=Truein SQLAlchemy prevents "connection already closed" errors after long idle periods.
Related conversions
Frequent conversions across the catalogue: