计算有条件的行数的混合表达式

Question

我有 4 张桌子：

class Cebola(Base):
    __tablename__ = 'cebolas'
    id = Column(Integer, primary_key=True, autoincrement=True)


class Checklist(Base):
    __tablename__ = 'checklists'

    id = Column(Integer, primary_key=True, autoincrement=True)
    checklist_type = Column(String)
    cebola_id = Column(Integer, ForeignKey('cebolas.id'))
    cebola = relationship('Cebola', backref=backref('checklists'))

    __mapper_args__ = {'polymorphic_on': checklist_type,
                       'polymorphic_identity': 'generic'}


class ChecklistA(Checklist):
    __tablename__ = 'checklist_a'

    id = Column(Integer, ForeignKey('checklists.id', ondelete='CASCADE'), primary_key=True)
    notes = Column(Unicode)

    __mapper_args__ = {'polymorphic_identity': 'a'}


class ChecklistB(Checklist):
    __tablename__ = 'checklist_b'

    id = Column(Integer, ForeignKey('checklists.id', ondelete='CASCADE'), primary_key=True)
    notes = Column(Unicode)

    __mapper_args__ = {'polymorphic_identity': 'b'}

现在我需要一种方法（可能是混合属性），它可以告诉我在 notes <> '' 的 Cebola 中有多少清单。

我已添加：

class Cebola(Base):

    @hybrid_property
    def number_of_comments(self):
        return len([c for c in self.checklists if c.notes])

    @number_of_comments(cls)
    def number_of_comments(cls):
        ???

我在SQLAlchemy - Writing a hybrid method for child count 中发现了类似的问题，但我的示例稍微复杂一些。

Answer 1

以下内容应该可以帮助您入门：

class Cebola(Base):
    __tablename__ = 'cebolas'
    id = Column(Integer, primary_key=True, autoincrement=True)

    @hybrid_property
    def number_of_comments(self):
        # return len([c for c in self.checklists if c.notes])
        # @NOTE:1: below will prevent error for those that do not have `notes` column
        # @NOTE:2: in any case this may issue a lot of SQL statements to load all subclasses of checklists relationship, so is not really efficient
        return len([c for c in self.checklists
                    if hasattr(c, 'notes') and c.notes])

    @number_of_comments.expression
    def number_of_comments(cls):
        from sqlalchemy import select
        ce, cl, cla, clb = [
            _.__table__ for _ in (Cebola, Checklist, ChecklistA, ChecklistB)]
        s = (
            select([func.count(cl.c.id).label("number_of_comments")])
            .select_from(cl.outerjoin(cla).outerjoin(clb))
            .where(or_(cla.c.notes != None, clb.c.notes != None))
            .where(cl.c.cebola_id == cls.id)
            .correlate(cls)
            .as_scalar()
        )
        return s

在查询中的用法：

q = session.query(Cebola, Cebola.number_of_comments)
for cebola, number_of_comments in q.all():
    print(cebola, number_of_comments)