如何实现自定义方法并将其与 SQLAlchemy 中的查询一起使用答案

【问题标题】：How to implement custom method and use it with query in SQLAlchemy如何实现自定义方法并将其与 SQLAlchemy 中的查询一起使用
【发布时间】：2017-06-28 05:34:33
【问题描述】：

我正在使用 SQLAlchemy，并且我有一个业务规则：“如果他们的所有 Bar 孩子都准备好了，Foo 就准备好了”，在一个用例中，我需要准备好所有 Foo，所以我正在尝试执行以下操作查询：

Session.query(Foo).filter(Foo.is_ready())

但我遇到了异常：

Traceback (most recent call last):
  File "/usr/local/lib/python3.5/dist-packages/sqlalchemy/orm/attributes.py", line 185, in __getattr__
    return getattr(self.comparator, key)
AttributeError: 'Comparator' object has no attribute 'all'

模型

class Foo(Base):
    bars = relationship(Bar, lazy = "dynamic")
    @classmethod
    def is_ready(self):
        return len(self.bar.all()) == len(self.bars.filter(Bar.status == "ready"))

class Bar(Base):
    status = Column(String)
    foo_id = Column(Integer, ForeignKey("foo.id"))

我做错了什么？我真的需要实现一个 Foo.is_ready() 方法，因为未来的业务规则会更复杂，所以封装该行为以便以后重用很重要

【问题讨论】：

标签： python sqlalchemy

【解决方案1】：

您的代码不起作用的原因是因为classmethod 中的self 是类本身，即Foo。（这就是为什么通常将其命名为cls 而不是self。）当然Foo.bars 没有.all()，因为Foo.bars 是关系本身，而不是Query 对象。

写这个的正确方法是什么？在这些情况下，将自己从 SQLAlchemy 的魔力中解脱出来并考虑需要编写的 SQL 会很有帮助。一个简单的方法是使用EXISTS：

SELECT * FROM foo
WHERE NOT EXISTS (
  SELECT * FROM bar
  WHERE bar.foo_id = foo.id AND bar.status != 'ready'
);

或JOIN：

SELECT * FROM foo
LEFT JOIN bar ON foo.id = bar.foo_id AND bar.status != 'ready'
WHERE bar.id IS NULL;

有了这个，现在写你的is_ready很容易：

class Foo(Base):
    @classmethod
    def is_ready(cls):
        return ~exists(select([Bar.id]).where(and_(Bar.foo_id == cls.id, Bar.status != "ready")))

session.query(Foo).filter(Foo.is_ready())

你甚至可以把它变成hybrid_property：

class Foo(Base):
    @hybrid_property
    def is_ready(self):
        return all(bar.status == "ready" for bar in self.bars)

    @is_ready.expression
    def is_ready(cls):
        bar = Bar.__table__
        return ~exists(select([Bar.id]).where(and_(Bar.foo_id == cls.id, Bar.status != "ready")))

session.query(Foo).filter(Foo.is_ready)

JOIN 很难像这样使用classmethod 或hybrid_property 来表达，所以你可以使用的一个技巧是.with_transformation：

class Foo(Base):
    @classmethod
    def is_ready(cls):
        def _transformer(query):
            return query.join(Bar, and_(Foo.id == Bar.foo_id, Bar.status != "ready")).filter(Bar.id.is_(None))
        return _transformer

session.query(Foo).with_transformation(Foo.is_ready())

【讨论】：

效果很好！但我还有一个问题。为什么is_ready(self) 是@hybrid_property？它可能是@property 的常规属性，如果我没记错的话，它也会很好用。
@Overflow012 hybrid_property 对实例和类都有效，因此您可以将Foo.is_ready 用作过滤器（使用表格上的列），以及在实例上使用foo.is_ready类的（使用加载在内存中的属性和关系）。

【解决方案2】：

没有bar 属性，而是bars。尝试使用hybrid_property 而不是classmethod。代码如下，但我没有测试过。

from sqlalchemy.ext.hybrid import hybrid_property

class Foo(Base):
    id = Column(Integer, primary_key=True)

    @hybrid_property
    def is_ready(self):
        return self.bars.count() == self.bars.filter_by(status="ready").count()

class Bar(Base): 
    id = Column(Integer, primary_key=True)
    status = Column(String)
    foo_id = Column(Integer, ForeignKey("foo.id"))
    foo = relationship(Foo, back_ref=back_ref("bars", lazy="dynamic"))

【讨论】：