SQLAlchemy如何通过循环实现动态多表关联查询功能

问题原因 #

你写的示例代码无法正常运行的核心问题有两个：

1. session.query() 要求传入多个位置参数代表要查询的表/字段，直接传入列表会被识别为单个列表参数，需要用*对列表做解包处理
2. 你当前场景的业务约定是所有辅助表都存在base_id字段和基础表的id字段关联，只要符合这个约定，循环关联的逻辑本身是成立的

可行实现方案 #

独立函数版本

from sqlalchemy.orm import Query

def merge_tables(base_table, list_of_ancillary_tables, session):
    # 解包表列表传入query
    query_tables = [base_table] + list_of_ancillary_tables
    stmt: Query = session.query(*query_tables)
    # 循环关联所有辅助表
    for ancillary_table in list_of_ancillary_tables:
        stmt = stmt.join(
            ancillary_table, 
            ancillary_table.base_id == base_table.id
            # 如果需要左连接，打开下面的注释即可
            # , isouter=True
        )
    return stmt

使用示例

# 初始化你的session_maker（和你原有逻辑一致即可）
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine("你的数据库连接串")
session_maker = sessionmaker(bind=engine)

with session_maker.begin() as session:
    # 场景1：关联全部3个辅助表
    stmt_full = merge_tables(BaseTable, [SupportTable1, SupportTable2, SupportTable3], session)
    results = stmt_full.all()

    # 场景2：只关联1个辅助表
    stmt_single = merge_tables(BaseTable, [SupportTable1], session)

    # 场景3：不关联任何辅助表
    stmt_base_only = merge_tables(BaseTable, [], session)

类内封装版本（适配你原有self.session_maker的写法）

class TableDao:
    def __init__(self, session_maker):
        self.session_maker = session_maker

    def merge_tables(self, base_table, list_of_ancillary_tables):
        with self.session_maker.begin() as session:
            query_tables = [base_table] + list_of_ancillary_tables
            stmt = session.query(*query_tables)
            for ancillary_table in list_of_ancillary_tables:
                stmt = stmt.join(ancillary_table, ancillary_table.base_id == base_table.id)
            return stmt.all()

拓展说明 #

如果后续辅助表的关联字段不统一，你可以额外传入关联条件映射参数，在循环里取对应表的关联条件即可，当前场景下以上实现完全可以满足需求。

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