Skip to content

Instantly share code, notes, and snippets.

@abe-winter
Last active March 30, 2023 21:39
Show Gist options
  • Save abe-winter/b89487c72310f79a8fd1e261f5ca5b67 to your computer and use it in GitHub Desktop.
Save abe-winter/b89487c72310f79a8fd1e261f5ca5b67 to your computer and use it in GitHub Desktop.
peewee migration

peewee migration notes

Peewee orm has built-in migration but it is not full-featured as of peewee 3.16.0.

It's missing:

  1. state tracking table that can track what's been applied
  2. auto-generate migrations
  3. the migrator doesn't have create table I think

These are my future-reference notes to myself documenting the minimal thing I built to solve the first problem above.

There are third-party libraries for generating + running peewee migrations; I have not checked them out, but if you are shopping, you should.

This isn't designed to be comprehensive or even reliable / safe, it's a minimal implementation to get a low-risk project into production.

import logging
from peewee import fn
from playhouse.migrate import SqliteMigrator, migrate
from models import db, SomeModel, MODELS, Migrations
logger = logging.getLogger(__name__)
migrator = SqliteMigrator(db)
def short_add(model, name):
"DRY migrator.add_column helper"
return migrator.add_column(model._meta.table_name, name, getattr(model, name))
# list of all migrations
migs = [
short_add(SomeModel, 'name'),
]
def run_missing_migrations():
db.connect()
schema_exists = Migrations.table_exists()
db.create_tables(MODELS, safe=True)
if not schema_exists:
logger.info('no migrations table, setting initial migration')
Migrations.insert(id=len(migs)).execute()
else:
max_index, = Migrations.select(fn.MAX(Migrations.id)).execute()
remaining_migs = migs[max_index.id:]
logger.info('max mig %d, total %d, applying %d', max_index.id, len(migs), len(remaining_migs))
migrate(*remaining_migs)
from peewee import SqliteDatabase, Model, DateTimeField, SQL, CharField
from playhouse.sqlite_ext import AutoIncrementField
db = SqliteDatabase('FILENAME.sqlite')
class Base(Model):
id = AutoIncrementField()
created = DateTimeField(constraints=[SQL('default current_timestamp')])
class Meta:
database = db
class Migrations(Base):
"this tracks migrations that have been applied"
pass
class SomeModel(Base):
name = CharField()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment