使用 SQLAlchemy，如何将行转换为“真正的”Python 对象？

Question

我一直在使用 SQLAlchemy 和 Alembic 来简化我使用的数据库访问，以及我对表所做的任何数据结构更改。这一直很有效，直到我开始注意到越来越多的 SQLAlchemy“过期”字段问题，从我的角度来看，几乎是随机的。

一个恰当的例子是这个片段，

class HRDecimal(Model):
    dec_id = Column(String(50), index=True)

    @staticmethod
    def qfilter(*filters):
        """
        :rtype : list[HRDecimal]
        """
        return list(HRDecimal.query.filter(*filters))


class Meta(Model):
    dec_id = Column(String(50), index=True)

    @staticmethod
    def qfilter(*filters):
        """
        :rtype : list[Meta]
        """
        return list(Meta.query.filter(*filters))

代码：

ids = ['1', '2', '3']  # obviously fake list of ids

decs = HRDecimal.qfilter(
    HRDecimal.dec_id.in_(ids))
metas = Meta.qfilter(
    Meta.dec_id.in_(ids))
combined = []

for ident in ids:
    combined.append((
        ident,
        [dec for dec in decs if dec.dec_id == ident],
        [hm for hm in metas if hm.dec_id == ident]
    ))

对于上述情况，没有问题，但是当我处理可能包含几千个 ID 的 ID 列表时，此过程开始花费大量时间，如果从 Flask 中的 Web 请求完成，则线程经常会被杀死。

当我开始探讨为什么会发生这种情况时，关键领域是

        [dec for dec in decs if dec.dec_id == ident],
        [hm for hm in metas if hm.dec_id == ident]

在组合这些（我认为是）Python 对象的某个时候，在某个时候调用dec.dec_id和hm.dec_id，在 SQLAlchemy 代码中，充其量，我们进入，

def __get__(self, instance, owner):
    if instance is None:
        return self

    dict_ = instance_dict(instance)
    if self._supports_population and self.key in dict_:
        return dict_[self.key]
    else:
        return self.impl.get(instance_state(instance), dict_)

作者InstrumentedAttribute在sqlalchemy/orm/attributes.py这似乎是很慢，但比这更糟糕的，我观察到时候过期的领域，然后我们进入，

def get(self, state, dict_, passive=PASSIVE_OFF):
    """Retrieve a value from the given object.
    If a callable is assembled on this object's attribute, and
    passive is False, the callable will be executed and the
    resulting value will be set as the new value for this attribute.
    """
    if self.key in dict_:
        return dict_[self.key]
    else:
        # if history present, don't load
        key = self.key
        if key not in state.committed_state or \
                state.committed_state[key] is NEVER_SET:
            if not passive & CALLABLES_OK:
                return PASSIVE_NO_RESULT

            if key in state.expired_attributes:
                value = state._load_expired(state, passive)

的AttributeImpl在同一个文件。这里可怕的问题是 state._load_expired 完全重新运行SQL 查询。所以在这样的情况下，有一个很大的 idents 列表，我们最终会对数据库运行数千个“小”SQL 查询，我认为我们应该只在顶部运行两个“大”查询。

现在，我通过如何使用 初始化烧瓶的数据库来解决过期问题session-options，更改

app = Flask(__name__)
CsrfProtect(app)
db = SQLAlchemy(app)

到

app = Flask(__name__)
CsrfProtect(app)
db = SQLAlchemy(
    app,
    session_options=dict(autoflush=False, autocommit=False, expire_on_commit=False))

这确实改善了上述情况，因为当行字段似乎随机过期时（根据我的观察），但访问 SQLAlchemy 项目的“正常”缓慢仍然是我们当前运行的问题。

SQLAlchemy 有什么办法可以从查询返回一个“真正的”Python 对象，而不是像现在这样的代理对象，所以它不受此影响？

Answer 1

您的随机性可能与在不方便的时间明确提交或回滚有关，或者由于某种自动提交。在其默认配置中，SQLAlchemy 会话在事务结束时使所有 ORM 管理的状态过期。这通常是一件好事，因为当事务结束时，您不知道数据库的当前状态是什么。这可以被禁用，就像你对expire_on_commit=False.

在ORM也不太适合在一般非常大的批量操作，如解释在这里。它非常适合处理复杂的对象图并将它们持久化到关系数据库中，而您无需付出太多努力，因为它会为您组织所需的插入等。其中一个重要部分是跟踪实例属性的更改。在SQLAlchemy的核心是更适合批量。

看起来您正在执行 2 个可能产生大量结果的查询，然后对数据进行手动“分组依据”，但效果不佳，因为对于每个 id，您都需要扫描整个结果列表，或O(nm)，其中 n 是 id 的数量，m 是结果。相反，您应该首先按 id 将结果分组到对象列表，然后执行“连接”。在其他一些数据库系统上，您可以直接在 SQL 中处理分组，但是 MySQL 没有数组的概念，除了 JSON。

例如，您的分组的性能​​可能更高的版本可能是：

from itertools import groupby
from operator import attrgetter

ids = ['1', '2', '3']  # obviously fake list of ids

# Order the results by `dec_id` for Python itertools.groupby. Cannot
# use your `qfilter()` method as it produces lists, not queries.
decs = HRDecimal.query.\
    filter(HRDecimal.dec_id.in_(ids)).\
    order_by(HRDecimal.dec_id).\
    all()

metas = Meta.query.\
    filter(Meta.dec_id.in_(ids)).\
    order_by(Meta.dec_id).\
    all()

key = attrgetter('dec_id')
decs_lookup = {dec_id: list(g) for dec_id, g in groupby(decs, key)}
metas_lookup = {dec_id: list(g) for dec_id, g in groupby(metas, key)}

combined = [(ident,
             decs_lookup.get(ident, []),
             metas_lookup.get(ident, []))
            for ident in ids]

请注意，由于在此版本中我们只对查询进行一次迭代，这all()不是绝对必要的，但也不应该受到太大影响。分组也可以在不使用 SQL 排序的情况下完成defaultdict(list)：

from collections import defaultdict

decs = HRDecimal.query.filter(HRDecimal.dec_id.in_(ids)).all()
metas = Meta.query.filter(Meta.dec_id.in_(ids)).all()

decs_lookup = defaultdict(list)
metas_lookup = defaultdict(list)

for d in decs:
    decs_lookup[d.dec_id].append(d)

for m in metas:
    metas_lookup[m.dec_id].append(m)

combined = [(ident, decs_lookup[ident], metas_lookup[ident])
            for ident in ids]

最后要回答您的问题，您可以通过查询 Core 表而不是 ORM 实体来获取“真正的”Python 对象：

decs = HRDecimal.query.\
    filter(HRDecimal.dec_id.in_(ids)).\
    with_entities(HRDecimal.__table__).\
    all()

这将产生一个类似namedtuple的对象列表，这些对象可以很容易地转换为 dict with _asdict()。