在SQLAlchemy中多列过滤时将空过滤值视为“任意值”的实现方案

1. 示例模型定义 #

先明确Car与Passenger的1:多关联模型（如果是多对多关系，后续会说明调整方式）：

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, Session
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Car(Base):
    __tablename__ = "cars"
    id = Column(Integer, primary_key=True)
    passengers = relationship("Passenger", back_populates="car")

class Passenger(Base):
    __tablename__ = "passengers"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)
    favorite_food = Column(String)
    car_id = Column(Integer, ForeignKey("cars.id"))
    car = relationship("Car", back_populates="passengers")

2. 核心问题分析 #

你之前用filter_non_null直接叠加过滤条件的方案，会要求所有条件匹配同一个关联乘客，这就是示例C失败的根本原因——找不到同时叫Allison（52岁）和Todd（17岁）的单个乘客。

正确思路是：每组独立的关联对象条件对应一个EXISTS子查询，确保存在至少一个关联对象满足该条件，再用and_组合这些子查询，实现「同时存在多个满足各自条件的关联对象」的逻辑。

3. 结合Pydantic动态生成查询 #

先定义支持多组乘客条件的Pydantic模型：

from pydantic import BaseModel, Field
from typing import List, Optional

class PassengerSearchCondition(BaseModel):
    name: Optional[str] = None
    age: Optional[int] = None
    favorite_food: Optional[str] = None

class CarSearchRequest(BaseModel):
    passenger_conditions: List[PassengerSearchCondition] = Field(..., description="多组乘客条件，需同时满足每组条件对应至少一个乘客存在")

然后编写动态查询生成函数：

from sqlalchemy import exists, and_
from sqlalchemy.orm import aliased

def search_cars(db: Session, request: CarSearchRequest):
    query = db.query(Car.id)
    
    for cond in request.passenger_conditions:
        # 为每组条件创建独立的Passenger别名，避免关联冲突
        PassengerAlias = aliased(Passenger)
        # 生成当前条件的过滤规则（仅保留非空字段）
        filters = []
        if cond.name:
            filters.append(PassengerAlias.name == cond.name)
        if cond.age:
            filters.append(PassengerAlias.age == cond.age)
        if cond.favorite_food:
            filters.append(PassengerAlias.favorite_food == cond.favorite_food)
        
        if filters:
            # 生成EXISTS子查询：存在属于当前Car的乘客满足该组条件
            subquery = exists().where(
                and_(
                    PassengerAlias.car_id == Car.id,
                    *filters
                )
            )
            query = query.filter(subquery)
    
    # 去重（同一Car可能匹配多个条件，避免结果重复）
    return [car.id for car in query.distinct().all()]

4. 验证示例查询 #

示例A：查找至少有一名乘客喜爱spaghetti的Car.id #

request = CarSearchRequest(
    passenger_conditions=[PassengerSearchCondition(favorite_food="spaghetti")]
)
result = search_cars(db, request)  # 返回[1,2]，符合预期

示例B：查找同时有乘客Sue和Stephen的Car.id #

request = CarSearchRequest(
    passenger_conditions=[
        PassengerSearchCondition(name="Sue"),
        PassengerSearchCondition(name="Stephen")
    ]
)
result = search_cars(db, request)  # 返回[]，符合预期

示例C：查找同时有名为Allison（52岁）和Todd（17岁）乘客的Car.id #

request = CarSearchRequest(
    passenger_conditions=[
        PassengerSearchCondition(name="Allison", age=52),
        PassengerSearchCondition(name="Todd", age=17)
    ]
)
result = search_cars(db, request)  # 返回[1]，符合预期

5. 扩展到多对多关系 #

如果Car和Passenger是多对多关联（通过中间表car_passenger），只需调整EXISTS子查询的关联逻辑：

# 定义中间表
car_passenger = Table(
    "car_passenger",
    Base.metadata,
    Column("car_id", Integer, ForeignKey("cars.id")),
    Column("passenger_id", Integer, ForeignKey("passengers.id"))
)

# 调整模型关系
class Car(Base):
    __tablename__ = "cars"
    id = Column(Integer, primary_key=True)
    passengers = relationship("Passenger", secondary=car_passenger, back_populates="cars")

class Passenger(Base):
    __tablename__ = "passengers"
    id = Column(Integer, primary_key=True)
    # ...其他字段
    cars = relationship("Car", secondary=car_passenger, back_populates="passengers")

修改EXISTS子查询的条件：

subquery = exists().where(
    and_(
        car_passenger.c.car_id == Car.id,
        car_passenger.c.passenger_id == PassengerAlias.id,
        *filters
    )
)

6. 优化点 #

• 用aliased为每组条件创建独立的关联表别名，避免多EXISTS子查询之间的关联冲突。
• 始终添加distinct()去重，防止同一Car因匹配多个条件重复出现在结果中。
• 若需要支持「满足A组条件或B组条件」的逻辑，可将多个EXISTS子查询用or_组合，灵活调整查询逻辑。

内容的提问来源于stack exchange，提问作者trn_is