SQLAlchemy中ForeignKey的查询过滤器答案

【问题标题】：Query filter over ForeignKey in SQLAlchemySQLAlchemy中ForeignKey的查询过滤器
【发布时间】：2021-08-10 03:07:01
【问题描述】：

为了简化，我有两个表（ORM FastAPI）

class Object(Base):
    __tablename__ = "object"
    id = Column('id', Integer, Identity(start=1), primary_key=True)
    name = Column(VARCHAR2(30), unique=True, index=True)
    attributes = relationship("Attributes", backref="parent", cascade="all, delete", passive_deletes=True)
    
class Attributes(Base):
    __tablename__ = "attributes"    
    id = Column('id', Integer, Identity(start=1), primary_key=True)
    attribute = Column(VARCHAR2(200), index=True)
    value = Column(VARCHAR2(2000), index=True)
    parent_id = Column(Integer, ForeignKey("object.id",ondelete="CASCADE"), nullable=False)

一个对象可以有多个属性，（1-N 关系）。属性是动态的（取决于对象，有些对象有 10 个属性，或者 50 个……）

例如：

Object      | Attributes
---------------------------------
Object 1    | color = red
            | form = round
            | level = 5
            | ...
            | attribute alpha
---------------------------------
Object 2    | color = red
            | form = square
            | level = 2
            | ...
            | attribute beta

我想做类似的事情：

“找到所有attribute.color = red且attribute.level >= 2且attribute.X被定义的对象”

我试过了：

query = db.query(Object).options( 
 joinedload(Attributes,innerjoin=False)).join(Attributes)
query = query.filter(Attributes.attribute == 'color')
query = query.filter(Attributes.value == 'red')
...
return query.all()

但我不知道如何在表属性上级联过滤器..

感谢您的帮助...

【问题讨论】：

标签： python sqlalchemy orm fastapi

【解决方案1】：

要实现过滤器，我会使用any()：

query = (
    session.query(Object)
    # -- NOTE: below join is not needed for the filter part;
    # .options(joinedload(Attributes, innerjoin=False))
    # .join(Attributes)
    # add additional criteria
    .filter(
        Object.attributes.any(
            and_(
                Attributes.attribute == "color",
                Attributes.value == "red",
            )
        )
    )
    .filter(
        Object.attributes.any(
            and_(
                Attributes.attribute == "level",
                func.cast(Attributes.value, Integer) >= 2,
            )
        )
    )
    .filter(Object.attributes.any(Attributes.attribute == "X"))  # exists
)

将产生 SQL 语句（具体取决于数据库引擎）：

SELECT object.id,
       object.name
FROM object
WHERE (EXISTS
         (SELECT 1
          FROM attributes
          WHERE object.id = attributes.parent_id
            AND attributes.attribute = 'color'
            AND attributes.value = 'red'))
  AND (EXISTS
         (SELECT 1
          FROM attributes
          WHERE object.id = attributes.parent_id
            AND attributes.attribute = 'level'
            AND CAST(attributes.value AS INTEGER) >= 2))
  AND (EXISTS
         (SELECT 1
          FROM attributes
          WHERE object.id = attributes.parent_id
            AND attributes.attribute = 'X'))

【讨论】：