peewee-migrations

Peewee Migrations & SQLite Schema Patterns

> **Scope**: Playhouse migrate operations, SQLite ALTER TABLE limitations, data migrations, and rollback procedures. > **Version range**: Peewee 3.x playhouse.migrate, SQLite 3.25+ (window functions), SQLite 3.35+ (DROP COLUMN) > **Generated**: 2026-04-14 — verify SQLite version availability in target deployment

---

Overview

Before SQLite 3.35, only `ADD COLUMN` and `RENAME TABLE` were available. All other schema changes require table rebuild. `playhouse.migrate` handles this correctly. Manual `execute_sql()` schema changes cause environment drift and broken rollbacks.

---

Pattern Table

| Operation | SQLite Support | Playhouse Method | Notes | |-----------|---------------|-----------------|-------| | Add column | All versions | `add_column()` | NULL default required for existing rows | | Rename column | 3.25+ | `rename_column()` | Before 3.25: table rebuild | | Drop column | 3.35+ | `drop_column()` | Before 3.35: table rebuild required | | Add index | All versions | `add_index()` | Non-blocking in SQLite | | Drop index | All versions | `drop_index()` | By index name, not column | | Add NOT NULL | Never directly | Table rebuild | SQLite can't modify column constraints | | Change column type | Never directly | Table rebuild | SQLite ignores type affinity changes | | Add FK constraint | Never directly | Table rebuild | FK constraints are table-level in SQLite |

---

Correct Patterns

Standard Column Addition

Add a nullable column first, then backfill, then add constraints in a separate migration.

from playhouse.migrate import SqliteMigrator, migrate
from peewee import TextField, IntegerField

def run_migration(db):
    migrator = SqliteMigrator(db)

    with db.atomic():
        migrate(
            # NULL required — existing rows cannot satisfy NOT NULL without backfill
            migrator.add_column('user', 'bio', TextField(null=True)),
            migrator.add_column('post', 'view_count', IntegerField(default=0)),
        )

    # Backfill after schema change, within same transaction if small dataset
    with db.atomic():
        db.execute_sql("UPDATE user SET bio = '' WHERE bio IS NULL")

**Why**: SQLite requires existing rows to satisfy the column's default. NOT NULL without default fails on non-empty tables. Two-step (add nullable, backfill, constrain) is the safe pattern.

---

Table Rebuild for Unsupported Changes

For changes SQLite can't do directly (column type, NOT NULL removal, FK addition), use explicit table rebuild:

def rebuild_table_with_new_schema(db):
    """Manual table rebuild when ALTER TABLE can't handle the change."""
    with db.atomic():
        # Step 1: Create new table with desired schema
        db.execute_sql('''
            CREATE TABLE user_new (
                id INTEGER PRIMARY KEY,
                username TEXT NOT NULL UNIQUE,
                email TEXT NOT NULL,          -- Changed from nullable
                created_at REAL NOT NULL       -- Changed from INTEGER
            )
        ''')

        # Step 2: Copy data with any needed transforms
        db.execute_sql('''
            INSERT INTO user_new (id, username, email, created_at)
            SELECT id, username,
                   COALESCE(email, 'unknown@example.com'),  -- Backfill nulls
                   CAST(created_at AS REAL)
            FROM user
        ''')

        # Step 3: Drop old table
        db.execute_sql('DROP TABLE user')

        # Step 4: Rename new table
        db.execute_sql('ALTER TABLE user_new RENAME TO user')

        # Step 5: Recreate indexes (dropped with old table)
        db.execute_sql('CREATE INDEX idx_user_email ON user(email)')

**Why**: SQLite ALTER TABLE cannot change types, remove NOT NULL, or add FK constraints. Rebuild copies data with transforms, atomically swaps. All steps in one `db.atomic()` block.

---

Tracking Migration State

Simple version table prevents re-running migrations:

from peewee import Model, CharField, DateTimeField
from datetime import datetime

class Migration(Model):
    name = CharField(unique=True)
    applied_at = DateTimeField(default=datetime.utcnow)

    class Meta:
        database = db

def has_run(name: str) -> bool:
    return Migration.select().where(Migration.name == name).exists()

def mark_run(name: str):
    Migration.create(name=name)

def run_migrations(db):
    Migration.create_table(safe=True)

    if not has_run('001_add_email_to_user'):
        migrator = SqliteMigrator(db)
        with db.atomic():
            migrate(migrator.add_column('user', 'email', TextField(null=True)))
            mark_run('001_add_email_to_user')

    if not has_run('002_add_post_index'):
        migrator = SqliteMigrator(db)
        with db.atomic():
            migrate(migrator.add_index('post', ('user_id', 'created_at'), False))
            mark_run('002_add_post_index')

---

Data Migration with Progress Tracking

Batch updates avoid long write locks:

def backfill_in_batches(db, batch_size=1000):
    """Backfill with batches to avoid long-running write locks."""
    offset = 0
    total = db.execute_sql('SELECT COUNT(*) FROM post WHERE slug IS NULL').fetchone()[0]

    while offset < total:
        with db.atomic():
            rows = db.execute_sql(
                'SELECT id, title FROM post WHERE slug IS NULL LIMIT ? OFFSET ?',
                (batch_size, offset)
            ).fetchall()

            if not rows:
                break

            for row_id, title in rows:
                slug = title.lower().replace(' ', '-')
                db.execute_sql('UPDATE post SET slug = ? WHERE id = ?', (slug, row_id))

        offset += batch_size
        print(f'Backfilled {min(offset, total)}/{total} rows')

**Why**: Long write transactions block other writes (WAL) or all reads AND writes (default journal). Batching limits each lock window.

---

Pattern Catalog

<!-- no-pair-required: