Skip to main content

SQLAlchemy migrations

Project description

mig - SQLAlchemy migrations

mig ([M]ediaGoblin [i]s [G]reat!) was first written by Christopher Allan Webber for GNU MediaGoblin.

Since then, Joar Wandborg has extracted the essentials of the functionality from MediaGoblin and into a separate package which’s README you are currently reading.

Init migrations

Either run, name, models, migrations) or add the mig.models.MigrationData table manually.

If you already have a populated database you’ll need to create a MigrationData(name='migrations_handle', version=0) row for your migrations in the MigrationData table, otherwise mig will try to initiate the database.

Example setup

Create migration

from mig import RegisterMigration

from sqlalchemy import MetaData, Table, Column, Integer, Unicode, DateTime, \


@RegisterMigration(1, MIGRATIONS)
def create_site_table(db_conn):
    metadata = MetaData(bind=db_conn.bind)

    user_table = Table('user', metadata, autoload=True,

    site_table = Table('site', metadata,
            Column('id', Integer, primary_key=True),
            Column('domain', Unicode),
            Column('owner_id', Integer, ForeignKey(user_table.columns['id'])))



@RegisterMigration(2, MIGRATIONS)
def item_add_site_id(db_conn):
    metadata = MetaData(bind=db_conn.bind)

    item_table = Table('item', metadata, autoload=True)
    site_table = Table('site', metadata, autoload=True)

    site_id_col = Column('site_id', Integer, ForeignKey(



Register models

import bcrypt

from datetime import datetime

from migrate import changeset

from talkatv import db

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(60), unique=True)
    email = db.Column(db.String(255), unique=True)
    password = db.Column(db.String(60))

    def __init__(self, username, email, password=None, openid=None):
        self.username = username = email

        if password:

        if openid:
            self.openid = openid

    def __repr__(self):
        return '<User {0}>'.format(self.username)

    def set_password(self, password):
        self.password = bcrypt.hashpw(password, bcrypt.gensalt())

    def check_password(self, password):
        return bcrypt.hashpw(password, self.password) == self.password

class OpenID(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    url = db.Column(db.String())
    created = db.Column(db.DateTime)

    user_id = db.Column(db.Integer, db.ForeignKey(''))
    user = db.relationship('User',
            backref=db.backref('openids', lazy='dynamic'))

    def __init__(self, user, url):
        self.created = datetime.utcnow()
        self.user = user
        self.url = url

class Item(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String())
    url = db.Column(db.String(), unique=True)
    created = db.Column(db.DateTime)

    site_id = db.Column(db.Integer, db.ForeignKey(''))
    site = db.relationship('Site',
            backref=db.backref('items', lazy='dynamic'))

    def __init__(self, url, title, site=None):
        if site:
   = site

        self.title = title
        self.url = url

        self.created = datetime.utcnow()

    def __repr__(self):
        return '<Item {0} ({1})>'.format(
       if else None)

    def as_dict(self):
        me = {
                'title': self.title,
                'url': self.url,
                'created': self.created.isoformat()}

        return me

class Site(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    created = db.Column(db.DateTime)
    domain = db.Column(db.String)

    owner_id = db.Column(db.Integer, db.ForeignKey(''))
    owner = db.relationship('User',
            backref=db.backref('sites', lazy='dynamic'))

    def __init__(self, owner, domain):
        self.owner = owner
        self.domain = domain

        self.created = datetime.utcnow()

    def __repr__(self):
        return '<Site {0} ({1})>'.format(

class Comment(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    created = db.Column(db.DateTime)
    text = db.Column(db.String())

    item_id = db.Column(db.Integer, db.ForeignKey(''))
    item = db.relationship('Item',
            backref=db.backref('comments', lazy='dynamic'))

    user_id = db.Column(db.Integer, db.ForeignKey(''))
    user = db.relationship('User',
            backref=db.backref('comments', lazy='dynamic'))

    def __init__(self, item, user, text):
        self.item = item
        self.user = user
        self.text = text

        self.created = datetime.utcnow()

    def __repr__(self):
        return '<Comment {0} ({1})>'.format(
                self.text[:25] + ('...' if len(self.text) > 25 else ''),

    def as_dict(self):
        me = {
                'username': self.user.username,
                'text': self.text,
                'created': self.created.isoformat()}
        return me


Run migrations

from mig import run
from mig.models import MigrationData

from yourapp import db
from yourapp.models import MODELS
from yourapp.migrations import MIGRATIONS

def check_or_create_mig_data():
    if not db.engine.dialect.has_table(db.session, 'mig__data'):
        # Create migration table

        # Create the first migration, so that mig doesn't init.
        migration = MigrationData(name=u'__main__', version=0)

if __name__ == '__main__':
    if db.engine.dialect.has_table(db.session, 'user'):
        # The DB is already populated, check if migrations are active,
        # otherwise create the migration data table

    run(db.engine, u'__main__', MODELS, MIGRATIONS)

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

mig-0.0.6.tar.gz (6.5 kB view hashes)

Uploaded Source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page