Python Database Patterns for Production

Moving from tutorial code to production requires patterns, not just knowledge. Production systems must handle schema evolution without downtime, monitor slow queries, backup data reliably, and organize code so it doesn't turn into a tangled mess. This tutorial covers architectural patterns, migration strategies, and debugging techniques that separate professional applications from hobby projects.

You'll learn the Data Access Object (DAO) pattern for encapsulating database logic, migrations for schema evolution, and how to monitor query performance in production systems.

The DAO (Data Access Object) Pattern

The DAO pattern isolates database logic from business logic. Instead of writing SQL directly in your views or business code, you create DAO classes that handle all database operations:

import sqlite3
from typing import Optional, List

class UserDAO:
    """Data Access Object for user records."""
    
    def __init__(self, db_path: str):
        self.db_path = db_path
    
    def create(self, name: str, email: str) -> int:
        """Create a user and return the ID."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            try:
                cursor.execute(
                    'INSERT INTO users (name, email) VALUES (?, ?)',
                    (name, email)
                )
                conn.commit()
                return cursor.lastrowid
            except sqlite3.IntegrityError as e:
                raise ValueError(f"User creation failed: {e}")
    
    def get_by_id(self, user_id: int) -> Optional[dict]:
        """Fetch a user by ID."""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute('SELECT id, name, email, created_at FROM users WHERE id = ?', (user_id,))
            row = cursor.fetchone()
            return dict(row) if row else None
    
    def get_all(self, limit: int = 100, offset: int = 0) -> List[dict]:
        """Fetch all users with pagination."""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute(
                'SELECT id, name, email, created_at FROM users ORDER BY created_at DESC LIMIT ? OFFSET ?',
                (limit, offset)
            )
            return [dict(row) for row in cursor.fetchall()]
    
    def update(self, user_id: int, **kwargs) -> bool:
        """Update user fields. Returns True if successful."""
        allowed_fields = {'name', 'email'}
        updates = {k: v for k, v in kwargs.items() if k in allowed_fields}
        
        if not updates:
            return True
        
        set_clause = ', '.join(f'{field} = ?' for field in updates.keys())
        values = list(updates.values()) + [user_id]
        
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            try:
                cursor.execute(f'UPDATE users SET {set_clause} WHERE id = ?', values)
                conn.commit()
                return cursor.rowcount > 0
            except sqlite3.IntegrityError as e:
                raise ValueError(f"Update failed: {e}")
    
    def delete(self, user_id: int) -> bool:
        """Delete a user. Returns True if successful."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
            conn.commit()
            return cursor.rowcount > 0

# Usage in your business logic
user_dao = UserDAO('/data/app.db')

# Create
user_id = user_dao.create('Alice', '[email protected]')
print(f"Created user {user_id}")

# Read
user = user_dao.get_by_id(user_id)
print(f"User: {user}")

# Update
user_dao.update(user_id, email='[email protected]')

# Delete
user_dao.delete(user_id)

The DAO pattern provides:

• Encapsulation: Database logic is hidden. Business code doesn't know SQL.
• Reusability: Multiple parts of your application share the same data access methods.
• Testability: You can mock the DAO for unit tests without touching the real database.
• Maintainability: Schema changes are isolated to the DAO; business logic is unaffected.

Using SQLAlchemy ORM for Production

SQLAlchemy is the industry-standard ORM for Python. It handles parameterization automatically, supports migrations, and provides a clean API:

from sqlalchemy import create_engine, Column, Integer, String, DateTime, func
from sqlalchemy.orm import declarative_base, Session
from sqlalchemy.exc import IntegrityError
from datetime import datetime

# Define the base class for ORM models
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)
    email = Column(String(255), unique=True, nullable=False)
    created_at = Column(DateTime, default=func.now())
    
    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"

# Create engine and session factory
engine = create_engine(
    'postgresql://user:password@localhost:5432/myapp',
    pool_size=10,
    max_overflow=20,
)

# Create tables if they don't exist
Base.metadata.create_all(engine)

# Session is used for query and commit operations
with Session(engine) as session:
    # Create
    user = User(name='Alice', email='[email protected]')
    session.add(user)
    session.commit()
    print(f"Created user {user.id}")
    
    # Read
    alice = session.query(User).filter_by(email='[email protected]').first()
    print(f"Found: {alice}")
    
    # Update
    alice.name = 'Alice Smith'
    session.commit()
    
    # Delete
    session.delete(alice)
    session.commit()

SQLAlchemy abstracts away the need to write raw SQL for CRUD operations, handles parameterization automatically, and integrates with web frameworks like Flask and FastAPI.

Database Migrations with Alembic

Schema changes in production require careful coordination—you can't just run ALTER TABLE. Alembic is Python's migration tool (like Django migrations or Rails migrations):

pip install alembic
alembic init migrations

Create a migration:

alembic revision --autogenerate -m "Add users table"

This generates a migration file in migrations/versions/:

# migrations/versions/001_add_users_table.py
from alembic import op
import sqlalchemy as sa

def upgrade():
    # Run when migrating forward
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('name', sa.String(255), nullable=False),
        sa.Column('email', sa.String(255), unique=True, nullable=False),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
        sa.PrimaryKeyConstraint('id')
    )

def downgrade():
    # Run when rolling back
    op.drop_table('users')

Apply migrations:

alembic upgrade head  # Apply all pending migrations
alembic downgrade -1  # Rollback one migration
alembic current       # Show current migration

Best practices for migrations:

1. Write migrations for all schema changes. Never run raw ALTER TABLE in production.
2. Test migrations on a staging database first. Verify they work before deploying.
3. Make migrations reversible. Implement upgrade() and downgrade().
4. One logical change per migration. "Add users table" and "Add index to users.email" are separate migrations.

Monitoring Query Performance

Slow queries are invisible until you measure them. Add query logging to catch problems:

import logging
import time
import sqlite3

# Configure logging to show slow queries
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

class LoggingConnection:
    """Wrapper around sqlite3 connection that logs slow queries."""
    
    def __init__(self, db_path: str, slow_threshold_ms: float = 100):
        self.conn = sqlite3.connect(db_path)
        self.slow_threshold_ms = slow_threshold_ms
    
    def cursor(self):
        return LoggingCursor(self.conn.cursor(), self.slow_threshold_ms)
    
    def commit(self):
        self.conn.commit()
    
    def rollback(self):
        self.conn.rollback()
    
    def close(self):
        self.conn.close()

class LoggingCursor:
    """Cursor wrapper that logs slow queries."""
    
    def __init__(self, cursor, slow_threshold_ms: float = 100):
        self.cursor = cursor
        self.slow_threshold_ms = slow_threshold_ms
    
    def execute(self, query: str, params=None):
        start = time.time()
        try:
            return self.cursor.execute(query, params or ())
        finally:
            elapsed_ms = (time.time() - start) * 1000
            if elapsed_ms > self.slow_threshold_ms:
                logger.warning(f"Slow query ({elapsed_ms:.1f}ms): {query}")

# Usage
conn = LoggingConnection('/data/app.db', slow_threshold_ms=50)
cursor = conn.cursor()

cursor.execute('SELECT * FROM users WHERE email = ?', ('[email protected]',))
# If this takes > 50ms, a warning is logged

conn.commit()
conn.close()

For PostgreSQL with psycopg, use the built-in logging:

import logging
import psycopg

# Enable query logging for psycopg
logging.basicConfig(level=logging.DEBUG)

conn = psycopg.connect(
    'postgresql://user:pass@localhost/myapp',
    # Uncomment to log all queries (verbose!)
    # logfile=sys.stderr
)

Backup and Recovery Strategies

Regular backups are non-negotiable. For SQLite3, copy the database file:

# One-time backup
cp app.db app.db.backup-2026-06-02

# Automated daily backup
0 2 * * * cp /data/app.db /backups/app.db.$(date +\%Y-\%m-\%d)

For PostgreSQL, use pg_dump:

# Full backup
pg_dump postgresql://user:password@localhost/myapp > myapp.sql

# Compressed backup (smaller)
pg_dump postgresql://user:password@localhost/myapp | gzip > myapp.sql.gz

# Scheduled daily backup
0 2 * * * pg_dump postgresql://user:password@localhost/myapp | gzip > /backups/myapp.$(date +\%Y-\%m-\%d).sql.gz

# Restore from backup
gunzip -c myapp.sql.gz | psql postgresql://user:password@localhost/myapp

Test recovery from backups regularly—a backup that doesn't restore is worthless.

Comparison: Patterns and Tools

Pattern	Complexity	Learning Curve	Best For
Raw SQL + DAO	Low	Low	Small to medium projects
SQLAlchemy ORM	Medium	Medium	Web applications, complex queries
Django ORM	Medium	Medium	Django applications
Raw SQL with psycopg	Low	Very low	Scripts, data pipelines

Key Takeaways

• Use the DAO pattern to isolate database logic and improve testability.
• SQLAlchemy is the industry standard ORM for Python; it handles parameterization automatically.
• Use Alembic for schema migrations; never run raw ALTER TABLE in production.
• Monitor slow queries with logging to catch performance bottlenecks before users report them.
• Backup regularly and test recovery procedures—automated backups that don't restore are worthless.
• One logical change per migration; keep migrations reversible.

Frequently Asked Questions

Should I use an ORM or write raw SQL?

ORMs are safer (automatic parameterization), more maintainable, and easier to test. Use an ORM (SQLAlchemy) unless you have a specific reason (extreme performance needs, complex database-specific features). Most production systems benefit from an ORM.

How often should I back up my database?

Depends on how much data loss is acceptable. For production systems, daily backups are standard. High-value systems back up hourly or use continuous replication.

Can I skip migrations and just run ALTER TABLE directly?

In development, yes. In production, never. Migrations let you roll back, document changes, and coordinate with deployments. Always use migrations.

How do I find slow queries in a production system?

Enable query logging (PostgreSQL's log_statement='all', SQLAlchemy's echo=True), monitor application metrics (response times), and use APM tools (Sentry, New Relic). Set a slow-query threshold (e.g., 100ms) and log anything slower.

Is it safe to use SQLAlchemy with SQLite3?

Yes, but SQLite3's concurrency limitations remain. SQLAlchemy doesn't magic away the fact that SQLite3 uses file locks. For high-concurrency applications, migrate to PostgreSQL.

How do I test database code?

Use an in-memory SQLite3 database for unit tests:

import sqlite3
from sqlalchemy import create_engine

# In-memory database for testing
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

# Now run your tests

Further Reading

• SQLAlchemy ORM Documentation
• Alembic Migration Tool
• PostgreSQL Backup and Restore
• Database Performance Tuning