如何在Python执行SQL查询时批量排除CSV值，避免大量占位符

我想用Python实现自动从SQL查询中排除CSV文件内的数百个值，目前已将CSV中的值转换为列表，通过cursor.execute(statement, exclusion())执行.sql文件中的查询语句，其中exclusion()返回该列表。但当前需在SQL语句的NOT IN子句中为列表每个值添加一个?占位符，而该列表每周都会更新，手动维护这些占位符非常繁琐，希望找到无需手动添加大量占位符的解决方法。

以下是我使用的完整代码：

# 将CSV中的值转换为列表
def exclusion():
    repeat_exclusion_df = pd.read_csv('repeat_names.csv')
    names_list = list(repeat_exclusion_df['names'])
    return names_list


# 读取.sql文件并转换为pandas DataFrame
import pandas as pd
import logging

cursor = connection_sql(uid=arguments.company_uid,
                        pwd=arguments.company_pwd,
                        server=arguments.company_server,
                        db=arguments.company_db)
with open('names_query.sql') as sql_file:
    logging.info('Open sql file')
    df_list = []
    for statement in sql_file.read().split(';'):
        db_cursor = cursor.execute(statement, exclusion())
        logging.info('Read each sql query')
        if db_cursor is not None:
            results = db_cursor.fetchall()
            if results:
                df = pd.DataFrame(results)
                df.columns = results[0].keys()
                df_list.append(df)

.sql文件中的查询示例：

SELECT NAME
FROM [DATABASE]
WHERE NAME NOT IN (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)

方法1：动态生成占位符字符串 #

直接在Python中根据排除列表的长度，自动生成对应数量的?占位符，替换SQL语句中固定的占位符块。

修改后的代码示例：

def exclusion():
    repeat_exclusion_df = pd.read_csv('repeat_names.csv')
    names_list = list(repeat_exclusion_df['names'])
    return names_list

import pandas as pd
import logging

# 获取排除列表
exclude_names = exclusion()
# 生成对应数量的占位符
placeholders = ', '.join(['?'] * len(exclude_names))

cursor = connection_sql(uid=arguments.company_uid,
                        pwd=arguments.company_pwd,
                        server=arguments.company_server,
                        db=arguments.company_db)
with open('names_query.sql') as sql_file:
    logging.info('Open sql file')
    df_list = []
    # 读取SQL内容，建议直接在sql文件中用标记占位，比如WHERE NAME NOT IN ({PLACEHOLDERS})
    sql_content = sql_file.read().replace('{PLACEHOLDERS}', placeholders)
    
    for statement in sql_content.split(';'):
        # 跳过空语句
        if not statement.strip():
            continue
        db_cursor = cursor.execute(statement, exclude_names)
        logging.info('Execute sql query')
        if db_cursor is not None:
            results = db_cursor.fetchall()
            if results:
                df = pd.DataFrame(results)
                df.columns = results[0].keys()
                df_list.append(df)

优化说明：把.sql文件中的NOT IN子句改成WHERE NAME NOT IN ({PLACEHOLDERS})，替换时更精准，避免正则匹配的复杂度。

方法2：使用临时表批量插入排除值 #

把CSV中的排除值插入到数据库的临时表中，然后在SQL查询中直接关联临时表，彻底摆脱占位符的限制。

代码示例：

def exclusion():
    repeat_exclusion_df = pd.read_csv('repeat_names.csv')
    names_list = list(repeat_exclusion_df['names'])
    return names_list

import pandas as pd
import logging

exclude_names = exclusion()
cursor = connection_sql(uid=arguments.company_uid,
                        pwd=arguments.company_pwd,
                        server=arguments.company_server,
                        db=arguments.company_db)

# 创建临时表
cursor.execute("CREATE TABLE #temp_exclusion (name NVARCHAR(255))")
# 批量插入排除值
insert_query = "INSERT INTO #temp_exclusion (name) VALUES (?)"
cursor.executemany(insert_query, [(name,) for name in exclude_names])

with open('names_query.sql') as sql_file:
    logging.info('Open sql file')
    df_list = []
    # 修改SQL语句，直接查询临时表
    sql_content = sql_file.read().replace(
        "WHERE NAME NOT IN (?,?,?,?)", 
        "WHERE NAME NOT IN (SELECT name FROM #temp_exclusion)"
    )
    for statement in sql_content.split(';'):
        if not statement.strip():
            continue
        db_cursor = cursor.execute(statement)
        logging.info('Execute sql query')
        if db_cursor is not None:
            results = db_cursor.fetchall()
            if results:
                df = pd.DataFrame(results)
                df.columns = results[0].keys()
                df_list.append(df)

# 临时表会在连接关闭后自动销毁，也可手动删除
cursor.execute("DROP TABLE #temp_exclusion")

方法3：使用表值参数（仅适用于SQL Server） #

如果你的数据库是SQL Server，可以定义自定义表类型，将Python中的列表作为表值参数传入SQL，这种方式性能更优，适合大量数据的场景。

步骤1：在SQL Server中创建表值类型：

CREATE TYPE NameListType AS TABLE (Name NVARCHAR(255))

步骤2：Python代码修改：

import pyodbc
import pandas as pd
import logging

def exclusion():
    repeat_exclusion_df = pd.read_csv('repeat_names.csv')
    return repeat_exclusion_df['names'].tolist()

exclude_names = exclusion()
# 创建支持表值参数的连接
conn_str = f"DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={arguments.company_server};DATABASE={arguments.company_db};UID={arguments.company_uid};PWD={arguments.company_pwd}"
conn = pyodbc.connect(conn_str)
cursor = conn.cursor()

# 将列表转换为表值参数格式
tvp = [(name,) for name in exclude_names]

with open('names_query.sql') as sql_file:
    # 修改SQL语句，直接传入表值参数
    sql_content = sql_file.read().replace(
        "WHERE NAME NOT IN (?,?,?,?)",
        "WHERE NAME NOT IN (SELECT Name FROM ?)"
    )
    # 传入表值参数执行查询
    db_cursor = cursor.execute(sql_content, (tvp,))
    results = db_cursor.fetchall()
    if results:
        df = pd.DataFrame(results)
        df.columns = results[0].keys()
        df_list = [df]

conn.commit()
conn.close()

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