使用SQLAlchemy传递Excel读取的列表作为SQL参数报错求助

问题原因 #

1. SQL语句拼接错误：直接用f-string将列表sg_items拼入IN ('{sg_items}')，会把列表转为字符串（比如"['something_1','something_2']"），导致SQL的IN子句识别为单个字符串值，而非多个参数，同时存在SQL注入风险。
2. params参数格式错误：pd.read_sql_query的params参数若为序列类型，要求是元组或字典的列表，直接传入普通列表不符合要求，且占位符数量与参数数量不匹配。

修正后的代码 #

import datetime
import pandas as pd
from datetime import date
from sqlalchemy import create_engine

def query_database(sg_items):
    end_date = date.today()
    start_date = date.today() - datetime.timedelta(days=90)

    # 生成IN子句对应的占位符（SQL Server使用?作为占位符）
    in_placeholders = ','.join(['?'] * len(sg_items))
    # 使用占位符构建SQL，避免字符串拼接错误和注入风险
    QUERY = """
     Select something,
            something_else,
            datetime
     from table
     where datetime between ? and ?
           and something IN ({in_placeholders})
     UNION
     Select something,
            something_else,
            datetime
     from other_table
     where datetime between ? and ?
           and something IN ({in_placeholders})
    """.format(in_placeholders=in_placeholders)

    # 构造参数元组：两次日期参数 + 两次列表参数（对应两个IN子句）
    params = (start_date, end_date) + tuple(sg_items) + (start_date, end_date) + tuple(sg_items)
    
    cnxn = create_engine("your_connection_string")  # 替换为实际连接字符串
    df = pd.read_sql_query(QUERY, cnxn, params=params)
    print(df)

def get_sg_list(input_file):
    # 简化列表读取，无需遍历行
    df = pd.read_excel(input_file)
    processed_data = df["Service Group"].tolist()
    print(processed_data)
    return processed_data

if __name__ == "__main__":
    input_file_name = "file.xlsx"
    output_file_name = "file.xlsx"

    sg_list = get_sg_list(input_file_name)
    query_database(sg_list)

关键说明 #

• 占位符处理：根据列表长度生成对应数量的?占位符，确保IN子句能正确接收多个参数。
• 参数传递：将日期和列表转为元组传入params，匹配SQL中所有占位符的顺序，同时符合SQLAlchemy的参数格式要求。
• 日期处理：直接传入date对象，由SQLAlchemy自动转换为SQL兼容的日期格式，避免手动字符串拼接的格式错误。
• 简化列表读取：使用df["Service Group"].tolist()替代iterrows遍历，效率更高且代码更简洁。

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