python - 将 "external' 类模型与 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 网络应用程序，我们在其中使用了相同的模型。我们发现了一个棘手的问题，因为 flask-sqlalchemy 的设计方式似乎是它期望模型中使用的所有类都通过传入 session 的事件实例来定义。下面是一个“正确的”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 session 。这把我们吓坏了，因为我们完全不知道如何将我们的 SqlAlchemy 模型集成到 Flask 中。我们真的特别想使用 flask-security 套件。

这个问题之前已经在SO上提出过。在这里，例如: How to use flask-sqlalchemy with existing sqlalchemy model?

我们的要求与那里接受响应的人的要求不同。响应指出失去了使用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)

为后代存档:

我花了很多时间寻找这个问题的答案。这比我最初问这个问题时要容易得多，但仍然不是很简单。

对于决定自己进行安全性的任何人，我推荐以下对使用 flask 的常见设计模式的出色阐述，但避免使用 flask-security 等不必要的依赖项: https://exploreflask.com/users.html

更新:对于任何感兴趣的人，与此相关的补丁已经开发了一段时间。截至目前它仍未发布，但您可以在此处查看其进度: https://github.com/mitsuhiko/flask-sqlalchemy/pull/250#issuecomment-77504337

更新:我从上面提到的补丁中获取了代码，并为 SQLAlchemy 对象创建了一个本地覆盖，它允许一个人注册一个外部基础。我认为这是最好的选择，直到 FSA 开始正式添加它。这是该类(class)的代码，供任何感兴趣的人使用。测试使用 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)