CODEWITHSUNDEEP
pythonsqlalchemy
Cristhian Boujon
Cristhian Boujon

Reputation: 4190

How to implement custom method and use it with query in SQLAlchemy

I'm using SQLAlchemy and I have the business rule: "Foo is ready if all their Bar children are ready" and in a use case I need to get all Foo that are ready so I'm trying to do the following query:

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

But I'm getting exception:

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'

models

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

What am I doing wrong? I really need to implement a method Foo.is_ready() since the business rule would be more complex in the future so it is important encapsulate that behavior in order to reutilize later

Upvotes: 3

Views: 5107

Answers (2)

univerio
univerio

Reputation: 20518

The reason your code doesn't work is because self in classmethod is the class itself, i.e. Foo. (This is why conventionally it is named cls instead of self.) Of course Foo.bars does not have .all(), because Foo.bars is the relationship itself, not a Query object.

What is the correct way to write this? In these scenarios, it is helpful to remove yourself from the magic of SQLAlchemy and think about the SQL that you need to write. An easy way is to use EXISTS:

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

Or a JOIN:

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

Armed with this, it's easy to now write your 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())

You can even turn it into a 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)

The JOIN is tricky to express using a classmethod or a hybrid_property like this, so one trick you can use is .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())

Upvotes: 4

hakubaa
hakubaa

Reputation: 169

There is no bar attribute, but bars. Try to use hybrid_property instead of classmethod. Code below, but I have not tested it.

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

Upvotes: 0

Related Questions