initial data for Flask model via fixture file.

base.py

from extensions import db

class Base(object):

    def save(self):
        db.session.add(self)
        db.session.commit()

data.json

[
    {
        "model":"models.test.EmailAddress",
        "fields":{
            "user_id":1,
            "email_address":"[email protected]"
        }
    }
]

manage.py

@manager.option('-f', '--fixture', default=None, help='Providing initial data for models Via fixture file')
def loaddata(fixture):
    if not os.path.isfile(fixture):
        print "please provoid a fixture file name"
    else:
        objects = get_fixture_objects(fixture)
        from sqlalchemy import create_engine
        
        del sys.modules['models']

        for obj in objects:
            models_arr = obj['model'].split(".")
            model_p = __import__('%s.%s'%(models_arr[0], models_arr[1]), globals(), locals(), [models_arr[2]])
            model = getattr(model_p, models_arr[2])

            try:
                bind = model.__bind_key__
            except AttributeError, e:
                bind = None
            
            if bind:
                engine = create_engine(app.config['SQLALCHEMY_BINDS'][bind])
            else:
                engine = create_engine(app.config['SQLALCHEMY_DATABASE_URI'])

            sql_temp = 'INSERT INTO "%s" (%s) VALUES (%s);'

            fields_arr = []
            values_arr = []

            for k in obj['fields']:
                fields_arr.append(k)
                if isinstance(obj['fields'][k],basestring):
                    values_arr.append("'%s'" % obj['fields'][k])
                else:
                    values_arr.append(obj['fields'][k])

            fields =  ','.join([str(s) for s in fields_arr])
            values =  ','.join([str(s) for s in values_arr])

            sql = sql_temp % (model.__tablename__, fields, values)
            print sql
            engine.execute(sql)

def get_fixture_objects(file):
    with open(file) as f:
        import json
        return json.loads(f.read())

test.py

from extensions import db
from .base import Base

class User(db.Model, Base):
    __tablename__ = 'user'
    __bind_key__ = 'test'

    id = db.Column(db.Integer, autoincrement=True, primary_key=True)
    username = db.Column(db.String(20), unique=True, nullable=False)

class Post(db.Model, Base):
    __tablename__ = 'post'
    __bind_key__ = 'test'

    id = db.Column(db.Integer, autoincrement=True, primary_key=True)
    title = db.Column(db.String(20), unique=True, nullable=False)

class EmailAddress(db.Model, Base):
    __tablename__ = 'addresses'
    __bind_key__ = 'test'

    id = db.Column(db.Integer, autoincrement=True, primary_key=True)
    user_id = db.Column(db.Integer)
    email_address = db.Column(db.String(20), unique=True, nullable=False)