Associate "external' Class Model With Flask Sqlalchemy

We use a central class model for a wide variety of python modules. This model is defined using SQLAlchemy. The classes all inherit from declarative_base. For example, our model de

Solution 1:

Answer :

As of today, the best way to do this is as follows:

Implement or import sqlalchemy base

from sqlalchemy.ext.declarative import declarative_base

base = declarative_base()


classBase(base):

    __abstract__ = True
    uid = Column(Integer, primary_key=True, autoincrement=True)

Register the external base:

from flask_sqlalchemy importSQLAlchemyfrom model.baseimportBase

app = Flask(__name__)
db = SQLAlchemy(app, model_class=Base)

Archived for posterity:

I spent a lot of time looking for an answer to this. This is a lot easier to do today than it was when I originally asked the question, but it still isn't exactly simple.

For anyone who decides to do security themselves, I recommend the following excellent exposition of common design patterns which use flask, but which avoid employing unnecessary dependencies like flask-security: https://exploreflask.com/users.html

UPDATE: For anyone interested, a patch has been in the works for some time related to this. As of now it still isn't released, but you can check its progress here: https://github.com/mitsuhiko/flask-sqlalchemy/pull/250#issuecomment-77504337

UPDATE: I have taken the code from the above mentioned patch and created a local override for the SQLAlchemy object which allows one to register an external base. I think this is the best option available until such time as FSA gets around to adding this officially. Here is the code from that class for anyone interested. Tested working with Flask-SqlAlchemy 2.2

Patching in register_external_base:

import flask_sqlalchemy
'''Created by Isaac Martin 2017. Licensed insofar as it can be according to the standard terms of the MIT license: https://en.wikipedia.org/wiki/MIT_License. The author accepts no liability for consequences resulting from the use of this software. '''classSQLAlchemy(flask_sqlalchemy.SQLAlchemy):
    def__init__(self, app=None, use_native_unicode=True, session_options=None,
                 metadata=None, query_class=flask_sqlalchemy.BaseQuery, model_class=flask_sqlalchemy.Model):

        self.use_native_unicode = use_native_unicode
        self.Query = query_classself.session = self.create_scoped_session(session_options)
        self.Model = self.make_declarative_base(model_class, metadata)
        self._engine_lock = flask_sqlalchemy.Lock()
        self.app = app
        flask_sqlalchemy._include_sqlalchemy(self, query_class)
        self.external_bases = []

        if app isnotNone:
            self.init_app(app)

    defget_tables_for_bind(self, bind=None):
        """Returns a list of all tables relevant for a bind."""
        result = []
        for Base in self.bases:
            for table in flask_sqlalchemy.itervalues(Base.metadata.tables):
                if table.info.get('bind_key') == bind:
                    result.append(table)

        return result

    defget_binds(self, app=None):
        """Returns a dictionary with a table->engine mapping.
        This is suitable for use of sessionmaker(binds=db.get_binds(app)).
        """
        app = self.get_app(app)
        binds = [None] + list(app.config.get('SQLALCHEMY_BINDS') or ())
        retval = {}
        for bind in binds:
            engine = self.get_engine(app, bind)
            tables = self.get_tables_for_bind(bind)
            retval.update(dict((table, engine) for table in tables))
        return retval

    @propertydefbases(self):
        return [self.Model] + self.external_bases

    defregister_base(self, Base):
        """Register an external raw SQLAlchemy declarative base.
        Allows usage of the base with our session management and
        adds convenience query property using self.Query by default."""

        self.external_bases.append(Base)
        for c in Base._decl_class_registry.values():
            ifisinstance(c, type):
                ifnothasattr(c, 'query') andnothasattr(c, 'query_class'):
                    c.query_class = self.Query
                ifnothasattr(c, 'query'):
                    c.query = flask_sqlalchemy._QueryProperty(self)

                    # for name in dir(c):#     attr = getattr(c, name)#     if type(attr) == orm.attributes.InstrumentedAttribute:#         if hasattr(attr.prop, 'query_class'):#             attr.prop.query_class = self.Query# if hasattr(c , 'rel_dynamic'):#     c.rel_dynamic.prop.query_class = self.Query

To be used like so:

app = Flask(__name__)
db = SQLAlchemy(app)
db.register_base(base)