Python flask.ext.sqlalchemy 模块，SQLAlchemy() 实例源码

def sql_select(explain=False):
    statement, params = load_query(request.args['query'])
    engine = SQLAlchemy().get_engine(current_app)

    if explain:
        if engine.driver == 'pysqlite':
            statement = 'EXPLAIN QUERY PLAN\n%s' % statement
        else:
            statement = 'EXPLAIN\n%s' % statement

    result = engine.execute(statement, params)
    return g.debug_toolbar.render('panels/sqlalchemy_select.html', {
        'result': result.fetchall(),
        'headers': result.keys(),
        'sql': format_sql(statement, params),
        'duration': float(request.args['duration']),
    })

def create_db(app):
    db = SQLAlchemy(app)
    return db

def __init__(self):
        self.db = SQLAlchemy()

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)
    mail.init_app(app)
    login_manager.setup_app(app)

    assets = Environment(app)
    assets.register(bundles())

    from .shared import shared as shared_blueprint
    app.register_blueprint(shared_blueprint)

    from .admin import admin as admin_blueprint
    app.register_blueprint(admin_blueprint)

    from .blog import blog as blog_blueprint
    app.register_blueprint(blog_blueprint)

    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)

    db.init_app(app)

    # with app.app_context():
    # # Extensions like Flask-SQLAlchemy now know what the "current" app
    # # is while within this block. Therefore, you can now run........
    #   db.create_all()

    return app

def db_setup():
    from app import context
    db = SQLAlchemy(app)
    context.setup(db)

def setup_user_roles():
    from app import context
    from flask.ext.sqlalchemy import SQLAlchemy
    db = SQLAlchemy()
    context.setup_user_roles()

def create_initial_db():
        from config import config
        from app.infrastructure.mappings import mapping
        from app.infrastructure.mappings.mapping import metadata
        db = SQLAlchemy()
        config = config[os.getenv('FLASK_CONFIG') or 'default']
        metadata.create_all(db.engine)

        if not os.path.exists(config.SQLALCHEMY_MIGRATE_REPO):
            api.create(config.SQLALCHEMY_MIGRATE_REPO, 'database repository')
            api.version_control(config.SQLALCHEMY_DATABASE_URI, config.SQLALCHEMY_MIGRATE_REPO)
        else:
            api.version_control(config.SQLALCHEMY_DATABASE_URI, config.SQLALCHEMY_MIGRATE_REPO, api.version(config.SQLALCHEMY_MIGRATE_REPO))

def authorize_changes(resource):
        return current_user.is_admin()

    # using SQLAlchemy's hybrid_property to provide a setter and validation step,
    # below I use Flask-Alcohol's setter decorator to do the same thing

def nav_title(self):
        return _('SQLAlchemy')

def title(self):
        return _('SQLAlchemy queries')