Flask-Alembic sans Flask-SqlAlchemy

db.py

from flask import g
from flask import current_app as app
from flask_alembic import Alembic

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


def setup_db_session():
    if not hasattr(g, 'db_session'):
        # DB connection added to request globals via Flask.before_request()
        engine = create_engine(app.config['SQLALCHEMY_DATABASE_URI'], convert_unicode=True)
        g.db_session = scoped_session(sessionmaker(autocommit=False,
                                                   autoflush=False,
                                                   bind=engine))
        Base.query = g.db_session.query_property()

    return g.db_session


def shutdown_db_session(exception=None):
    # DB connection shutdown called by Flask.teardown_appcontext()
    if hasattr(g, 'db_session'):
        g.db_session.remove()


class FlaskSqlAlchemyShim:
    """
    Stand-in for Flask-SqlAlchemy used by Flask-Alembic
    """
    class SqlAlchemy:
        def __init__(self, session):
            self.session = session

        @property
        def engine(self):
            return self.session.get_bind()

        @property
        def metadata(self):
            return Base.metadata

    @property
    def db(self):
        session = setup_db_session()
        return FlaskSqlAlchemyShim.SqlAlchemy(session)


def init_db(app):
    # create a fake Flask-SqlAlchemy extension for Flask-Alembic
    app.extensions['sqlalchemy'] = FlaskSqlAlchemyShim()

    # create SqlAlchemy session before request, and close afterwards
    app.before_request(setup_db_session)
    app.teardown_appcontext(shutdown_db_session)

    app.alembic = Alembic()
    app.alembic.init_app(app)