SQLite3 migrate by create and copy #SQLite #migrations

README.md

SQLite3 migrate by create and copy

• Place the source DB into data/old.db.
• Ensure your local DB (the one used for development) is up-to-date.
• Add the list of models and tables in dependent order.
• Add your own defaults for columns that are required but have no source data.
• Run the command, fix any problems.
• The db in data/new.db will have the latest schema and workable current data.

This does not need Flask to be used. It could be adapted for general use with SQLAlchemy only.

commands.py

from flask import Flask, current_app
from flask_script import Manager
from sqlalchemy import inspect, Table
from sqlalchemy.orm import Session

from example.models import db, Article, User, Publisher, article_publisher


app = Flask()

app.config['BINDS'] = {
        'old': 'sqlite3:///data/old.db',
        'new': 'sqlite3:///data/new.db',
}

manager = Manager(app)


def clone_row(row):
    """Need this because result row is not mutable"""

    new_row = {}
    for k in row.keys():
        new_row[k] = getattr(row, k)
    return new_row


@manager.command
def migrate():
    """Migrate tables by creating and copying data across."""

    """
    List the entities in dependent order.
    Can be either models or table objects.
    """
    entities = [
        User,
        Article,
        Publisher,
        article_publisher,
    ]

    """
    _old_ database is the one we are trying to migrate,
    and the source of (most of) our data.
    """
    old_engine = db.get_engine(current_app, 'old')
    old_session = Session(old_engine)

    "_new_ database is the destination we are filling."
    new_engine = db.get_engine(current_app, 'new')
    new_session = Session(new_engine)

    for entity in entities:
        "Tables are likely to be mostly M-M tables without a class definition."
        if type(entity) is Table:
            if entity.exists(bind=old_engine):
                "If the table exists in the old database, select the rows from it."
                rows = old_engine.execute(entity.select())
            else:
                "If it does not exist, select the rows from our primary (default) database."
                rows = db.engine.execute(entity.select())

            "Create the table in the new database."
            entity.create(bind=new_engine)

            "Insert each row into the newly created table."
            for row in rows:
                new_session.execute(entity.insert(row))
        else:
            "Here we are dealing with SQLalchemy models"
            if entity.__table__.exists(bind=old_engine):
                """
                If the table exists in the old DB, then inspect the old DB's
                columns and create a list of columns to be selected. This is
                needed because sometimes the new definition has new columns
                which do not exist on the old DB.
                """
                insp = inspect(old_engine)
                columns = [getattr(entity, c['name']) for c in insp.get_columns(entity.__tablename__) if hasattr(entity, c['name'])]

                "Creat dict clones of the results so we can mutate them later"
                rows = [clone_row(r) for r in old_session.query(*columns).all()]
            else:
                """
                If the table does not exist in the old DB, use the current
                (default) DB as a data source instead.
                """
                rows = db.session.query(entity).all()

            """
            Create a model which overrides the original entity, but binds to the
            new DB instead.
            """
            class NewModel(entity):
                __tablename__ = entity.__tablename__
                __bind_key__ = 'new'

            "Create the table in the new DB."
            entity.__table__.create(bind=new_engine)

            for row in rows:
                "We want to work with dictionaries, not objects."
                if hasattr(row, '__dict__'):
                    row = row.__dict__

                "Remove SQLAlchemy cruft if it's there."
                if '_sa_instance_state' in row:
                    del row['_sa_instance_state']

                """
                Add defaults for fields where the destination column
                requires a value, but there is no data in the source DB.
                """
                if entity.__tablename__ == 'users':
                    row['country'] = 'ZA'

                new_session.execute(entity.__table__.insert(row))

        new_session.commit()

if __name__ == '__main__':
    manager.run()    

Don't be tempted into making all the above branches work with tables. SQLAlchemy expects datetime objects when inserting (for example), which the model will give you, but a table select will not. The table select returns the value as a string instead.