在SQL执行中使用SQL Alchemy迭代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列表中对每个表进行查询并且change_set_ids仅为1int

但现在change_set_ids是ints的列表。

现在，对于每个change_set_id，我要遍历bmds_stage_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

考虑将表名格式化为原始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() 
)