使用 SQL Alchemy 在 SQL 执行中迭代 2 个列表

Question

我目前有这个查询：

        for table_name in bmds_stage_tables:
            get_updated_tables = conn.execute(
                f"""
                    DECLARE @csId INT = ?
                    SELECT * FROM bmds_stage.{table_name}
                    WHERE ChangeSetId = @csId
                """,
                change_set_ids
            ).fetchall()

我改用bindparams:

query_updated_tables = text(
    """
    SELECT * FROM bmds_stage.:table_name
    WHERE ChangeSetId in :csId
    """
).bindparams(
    bindparam("table_name"),
    bindparam("csId", expanding=True),
)

所以之前，我能够根据 bmds_stage_table 列表中的 table 进行查询，而 change_set_ids 只是 1 int

但现在 change_set_ids 是 int 的列表。

所以现在，对于每个 change_set_id，我想遍历 bmds_stage_table 列表

中的所有 table

随着更改，我不得不调整执行查询的方式：

        for table_name in bmds_stage_tables:
            get_updated_tables = conn.execute(query_updated_tables, {"table_name":table_name, "csId":change_set_id}).fetchall()

但是我得到这个错误：

sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) ('42000', "[42000] [Microsoft][ODBC Driver 11 for SQL Server][SQL Server]Incorrect syntax near '@P1'. (102) (SQLExecDirectW)")
[SQL: 
    SELECT * FROM bmds_stage.?
    WHERE ChangeSetId in (?)
    ]
[parameters: ('PM_Category', 1045)]

如有任何帮助，我将不胜感激！

Answer 1

考虑将 table 名称格式化为原始 SQL，因为标识符无法参数化，bind_params 用于文字值：

for table_name in bmds_stage_tables:
    query_updated_tables = text( 
        f"""SELECT * FROM bmds_stage.{table_name} 
           WHERE ChangeSetId in :csId 
        """ 
    ).bindparams( 
        bindparam("csId", expanding=True)
    )

    params = {"csId": change_set_id}
    get_updated_tables = (
        conn.execute(
            query_updated_tables, params
        ).fetchall() 
    )

也可能考虑 sqlalchemy 的函数形式：

from sqlalchemy.schema import Table
from sqlachemy import select, column, bindparam

for table_name in bmds_stage_tables:
   query_updated_tables = (
        select(
            Table(
                table_name,
                schema = "bmds_stage"
            )
        ).where(
            column("ChangeSetId").in_(
                bindparam("csId", expand=True)
            )
        )
    )

    params = {"csId": change_set_id}
    get_updated_tables = (
        conn.execute(
            query_updated_tables, params
        ).fetchall() 
    )

使用 SQL Alchemy 在 SQL 执行中迭代 2 个列表

Iterating over 2 lists in SQL execution using SQL Alchemy

python

sql

sql-server

sqlalchemy

python-3.x