将"外部"类模型与flask sqlalchemy相关联

Question

我们为各种python模块使用中央类模型.此模型使用SQLAlchemy定义.这些类都继承自declarative_base.

例如,我们的模型定义如下所示:

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'))

我们一直在构建一个烧瓶网络应用程序,我们在其中使用相同的模型.我们发现了一个棘手的问题,因为flask-sqlalchemy似乎是以这样的方式设计的,它希望通过传入会话的活动实例来定义其模型中使用的所有类.以下是"适当的"flask-sqalchemy类模型定义的示例:

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)

请注意,flask-sqlalchemy的上述示例需要已经实例化的sql会话.这让我们感到震惊,因为我们完全不知道如何将我们的SqlAlchemy模型集成到烧瓶中.我们真的想特别使用flask-security套件.

这个问题在SO之前提出过.这里,例如: 如何使用现有sqlalchemy模型的flask-sqlalchemy？

我们的要求与那些接受回复的人的要求不同.响应指出一个人失去了使用User.query的能力,但这正是我们必须保留的事情之一.

放弃我们漂亮,优雅,中心阶级的模型定义,而不是使用flask-sqlalchemy所需要的东西是不可行的.我们有什么方法可以将我们的模型与SQLAlchemy()对象相关联吗？在我们的类上获取.query()方法的加分点似乎是flask-security所需要的.

Answer 1

解:

截至今天,最好的方法如下:

实施或导入sqlalchemy base

from sqlalchemy.ext.declarative import declarative_base

base = declarative_base()


class Base(base):

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

注册外部基地:

from flask_sqlalchemy import SQLAlchemy
from model.base import Base

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

存档为后代:

我花了很多时间寻找答案.今天这比我最初提出这个问题要容易得多,但它仍然不是很简单.

对于那些决定自己做安全的人,我推荐以下对使用烧瓶的常见设计模式的优秀说明,但避免使用不必要的依赖关系,如烧瓶安全性:https: //exploreflask.com/users.html

更新: 对于任何感兴趣的人来说,补丁已经有一段时间了.截至目前它仍然没有发布,但你可以在这里检查它的进展:https: //github.com/mitsuhiko/flask-sqlalchemy/pull/250#issuecomment-77504337

更新: 我从上面提到的补丁中获取了代码,并为SQLAlchemy对象创建了一个本地覆盖,允许用户注册外部基础.我认为这是最好的选择,直到FSA开始正式添加这个.对于任何感兴趣的人,这是该类的代码.测试使用Flask-SqlAlchemy 2.2

在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

要像这样使用:

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