Associated "external" class model with colloid sqlalchemy

Question

We use a central class model for a wide range of python modules. This model is defined using SQLAlchemy. All classes inherit from declarative_base.

For example, our model definitions look something like this:

Base = declarative_base() class Post(Base): __tablename__ = 'Posts' id = Column(INT, primary_key=True, autoincrement=True) body = Column(TEXT) timestamp = Column(TIMESTAMP) user_id = Column(INT, ForeignKey('Users.uid')) 

We are creating a flash web application in which we use the same model. We found a difficult problem in that the sqlalchemy bulb seems to be designed in such a way that it expects all classes used in its model to be defined by passing in the active instance of the session. The following is an example of determining the “correct” colaloid sqalchemy model:

 app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db' db = SQLAlchemy(app) class User(db.Model): id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True) email = db.Column(db.String(120), unique=True) 

Note that the above example for flask-sqlalchemy requires an already created sql session. This horrified us, because we absolutely do not understand how to integrate our SqlAlchemy model into a flask. We really want to use the flash security package in particular.

This issue was raised earlier on SO. Here, for example: How to use collon-sqlalchemy with an existing sqlalchemy model?

Our requirements are different from those who accepted the answer. The response states that the user is losing the ability to use User.query, but that’s exactly what we have to save.

It is impossible to abandon our pretty, elegant, definition of a central class model in favor of what the jar-sqlalchemy seems to require. Is there a way to associate our model with an SQLAlchemy () object? Bonus points for getting the .query () method on our classes, which seems to be required using a flash drive.

Answer 1

I spent a lot of time finding the answer to this question. This is much easier to do today than when I initially asked this question, but it is still not entirely simple.

For those who decide to ensure their own security, I recommend the following excellent presentation of common design patterns that use a flask, but which avoid unnecessary dependencies such as flash protection: https://exploreflask.com/users.html

UPDATE: For anyone interested, the patch has been in work for some time related to this. It has not been released yet, but you can check its progress here: https://github.com/mitsuhiko/flask-sqlalchemy/pull/250#issuecomment-77504337

UPDATE: I took the code from the above patch and created a local override for the SQLAlchemy object, which allows you to register an external database. I think this is the best option available until the FSA starts adding this officially. Here is the code for this class for anyone interested. Tested using Flask-SqlAlchemy 2.2

Patch 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 

Used like this:

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