如何在Flask-SQLAlchemy中对父子类进行多条件组合过滤？

我定义了以下SQLAlchemy模型：

Pairing 模型 #

class Pairing(Base):
    __tablename__ = 'pairing'
    id: Mapped[int] = mapped_column(Integer, unique=True, primary_key=True, autoincrement=True)
    pairing_no: Mapped[str] = mapped_column(String(10), unique=False, nullable=False)
    total_expenses: Mapped[str] = mapped_column(String(10), unique=False, nullable=True)
    pairing_tafb: Mapped[str] = mapped_column(String(10), unique=False, nullable=True)

    flights: Mapped[List["Flight"]] = relationship(
        back_populates="pairing", cascade='all, delete'
    )

Flight 模型 #

class Flight(Base):
    __tablename__ = 'flight'
    id: Mapped[int] = mapped_column(Integer, unique=True, primary_key=True, autoincrement=True)
    destination_station: Mapped[str] = mapped_column(String, unique=False)

    pairing_no: Mapped[str] = mapped_column(ForeignKey('pairing.pairing_no'), unique=False)
    pairing: Mapped["Pairing"] = relationship(back_populates='flights')

目前已实现对Pairing的多条件过滤：

pairing_filter = {
        'pairing_no': request.form.get('search_pairing'),
        'pairing_tafb': request.form.get('pairing_tafb'),
        'total_expenses': request.form.get('total_expenses'),
    }
pairing_filter = {key: value for (key, value) in pairing_filter.items() if value}
pairings_to_display = db.session.scalars(select(Pairing).filter_by(**pairing_filter).order_by(Pairing.id)).all()

现在需要添加针对Flight的过滤条件（例如筛选包含目的地为LHR航班的Pairing），且支持用户输入的任意组合（如同时按pairing_no和Flight目的地过滤、仅按Flight目的地过滤等），请问该如何实现？

要实现跨关联模型的组合过滤，推荐用EXISTS子查询（性能更优，逻辑清晰）或JOIN+DISTINCT两种方式，以下是具体实现：

方法一：EXISTS子查询（推荐） #

这种方式会检查每个Pairing是否存在符合条件的关联Flight，不会返回重复的Pairing记录。

1. 获取Flight过滤参数 #

从请求中提取Flight相关的过滤值：

flight_destination = request.form.get('flight_destination')
# 若需扩展其他Flight字段，比如航班号，可继续添加：
# flight_number = request.form.get('flight_number')

2. 动态组合过滤条件 #

初始化查询对象后，逐步添加有值的过滤条件：

# 基础查询
query = select(Pairing).order_by(Pairing.id)

# 添加Pairing自身的过滤条件
if pairing_filter:
    query = query.filter_by(**pairing_filter)

# 添加Flight关联过滤条件（仅当用户输入了对应参数时）
if flight_destination:
    # 构造子查询：判断当前Pairing是否存在符合目的地的Flight
    subquery = select(Flight.id).where(
        Flight.pairing_no == Pairing.pairing_no,
        Flight.destination_station == flight_destination
    ).exists()
    query = query.filter(subquery)

# 执行查询
pairings_to_display = db.session.scalars(query).all()

扩展多Flight条件 #

如果需要同时过滤多个Flight字段，直接在子查询的where中追加条件即可：

if flight_destination and flight_number:
    subquery = select(Flight.id).where(
        Flight.pairing_no == Pairing.pairing_no,
        Flight.destination_station == flight_destination,
        Flight.flight_no == flight_number
    ).exists()
    query = query.filter(subquery)

方法二：JOIN+DISTINCT #

若偏好JOIN写法，可使用此方式，但需添加distinct()避免因一个Pairing对应多个Flight而返回重复记录：

# 基础查询
query = select(Pairing).order_by(Pairing.id)

# 添加Pairing过滤条件
if pairing_filter:
    query = query.filter_by(**pairing_filter)

# 添加Flight过滤条件
if flight_destination:
    query = query.join(Pairing.flights).filter(Flight.destination_station == flight_destination).distinct()

# 执行查询
pairings_to_display = db.session.scalars(query).all()

两种方法对比 #

• EXISTS：适合数据量较大的场景，数据库优化器更容易生成高效执行计划，避免重复数据。
• JOIN+DISTINCT：写法更直观，但数据量大时性能可能不如EXISTS，且需手动处理去重。

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