12

我们为各种 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 模型集成到烧瓶中。我们真的特别想使用烧瓶安全套件。

之前在 SO 上已经提出过这个问题。在这里，例如： 如何将 flask-sqlalchemy 与现有的 sqlalchemy 模型一起使用？

我们的要求与那里接受响应的人不同。响应指出失去了使用 User.query 的能力，但这正是我们必须保留的内容之一。

放弃我们漂亮的、优雅的、中心类模型定义来支持 flask-sqlalchemy 似乎需要的东西是不可行的。我们有什么方法可以将我们的模型与 SQLAlchemy() 对象关联起来？在我们的类上获得 .query() 方法的奖励积分似乎是烧瓶安全所必需的。

4

1 回答 1

19

解决方案：

截至今天，最好的方法如下：

实施或导入 sqlalchemybase

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)
于 2015-03-03T19:10:56.727 回答