能否复用SQLAlchemy的join()和where()语句用于查询与更新？

当然可以把重复的join()和where()逻辑打包复用，核心思路是把通用的关联过滤逻辑封装成可复用的函数，让查询和更新语句共享同一套逻辑，彻底消除冗余和不同步的风险。下面是具体实现方案：

方案1：封装通用过滤函数 #

定义一个函数，接收SQLAlchemy的语句对象（不管是Select还是Update），在函数内部统一执行所有重复的join和where操作，最后返回处理后的语句对象。这样查询和更新都只需要调用这个函数即可。

示例代码：

def apply_common_filters(statement):
    # 在这里统一写所有重复的join和where逻辑
    return (
        statement
        .join(other_table1, mytable.c.id == other_table1.c.mytable_id)
        .join(other_table2, other_table1.c.id == other_table2.c.other1_id)
        .join(other_table3, mytable.c.other3_id == other_table3.c.id)
        .where(mytable.c.status == "active")
        .where(other_table2.c.type == "target")
        .where(other_table3.c.date >= date(2024, 1, 1))
    )

# 查询时使用
query = apply_common_filters(session.query(func.sum(mytable.c.amount)))
sum_amount = session.execute(query).scalars().one()
assert sum_amount == 123, sum_amount

# 更新时使用
my_update = apply_common_filters(
    mytable.update().values(status="processed")
)
session.execute(my_update)

方案2：拆分复用关联与条件 #

如果需要更细粒度的复用，可以把关联逻辑和过滤条件分开定义，按需组合：

# 提前定义通用过滤条件
common_conditions = (
    (mytable.c.status == "active") &
    (other_table2.c.type == "target") &
    (other_table3.c.date >= date(2024, 1, 1))
)

def apply_common_joins(statement):
    return (
        statement
        .join(other_table1, mytable.c.id == other_table1.c.mytable_id)
        .join(other_table2, other_table1.c.id == other_table2.c.other1_id)
        .join(other_table3, mytable.c.other3_id == other_table3.c.id)
    )

# 查询
query = apply_common_joins(session.query(func.sum(mytable.c.amount))).where(common_conditions)
# 更新
my_update = apply_common_joins(mytable.update().values(status="processed")).where(common_conditions)

这两种方式都能保证关联和过滤逻辑只维护一份，修改时只需调整函数或条件变量，查询和更新会自动同步，完全避免人为遗漏的问题。

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