Associate "external' class model with flask sqlalchemy Associate "external' class model with flask sqlalchemy flask flask

Associate "external' class model with flask sqlalchemy


Solution:

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

Implement or import sqlalchemy base

from sqlalchemy.ext.declarative import declarative_basebase = declarative_base()class Base(base):    __abstract__ = True    uid = Column(Integer, primary_key=True, autoincrement=True)

Register the external base:

from flask_sqlalchemy import SQLAlchemyfrom model.base import Baseapp = 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. '''class SQLAlchemy(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_class        self.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 is not None:            self.init_app(app)    def get_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    def get_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    @property    def bases(self):        return [self.Model] + self.external_bases    def register_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():            if isinstance(c, type):                if not hasattr(c, 'query') and not hasattr(c, 'query_class'):                    c.query_class = self.Query                if not hasattr(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)