Python pandas+pyodbc参数化SQL查询报错解决方案

问题复现 #

需求为编写支持传入动态参数的函数，循环执行结构一致的不同SQL查询，初始实现代码如下：

import pandas as pd
cnxn_str = ("Driver={SQL Server Native Client 10.0};"
            "Server=xx.xxx.xxx.xx,xx;"
            "Database=dbase;"
            "UID=sa;"
            "PWD=pswd;")
cnxn = pyodbc.connect(cnxn_str)

cursor = cnxn.cursor()

def TRX(tbl_name, var_value):
  import pandas as pd
  #route by hour----
  query = print("SELECT TOP 100 * FROM [DB].[dbo]." + tbl_name + " WHERE var1 IN ('" + var_value + "')") 
  df = pd.read_sql_query(query, cnxn)
  return(df)

使用如下方式调用函数：

TRX(tbl_name = '[table1]', var_value = 'a')

程序抛出错误：

DatabaseError: Execution failed on sql 'None': The first argument to execute must be a string or unicode query.

错误根因 #

• 直接触发报错的原因：print()函数的返回值为None，代码将print()的执行结果赋值给query变量，最终传入pd.read_sql_query的SQL参数是None，不符合接口要求的字符串类型。
• 存在安全与稳定性隐患：直接通过字符串拼接生成SQL语句，一是存在SQL注入风险，二是当传入参数包含单引号等特殊字符时会直接触发SQL语法错误。
• 隐性代码问题：代码中使用了pyodbc但未在头部导入该库，实际运行会先触发模块未找到的错误；函数内部重复导入pandas属于冗余操作。

正确实现方案 #

SQL语法中，表名、列名属于标识符，无法通过数据库驱动的参数化接口传值，需要通过白名单校验+标识符转义的方式规避注入风险；查询条件中的值必须使用驱动支持的参数占位符传入，pyodbc对应的占位符为?。
修正后的可运行代码如下：

import pandas as pd
import pyodbc
from typing import Union, List

# 初始化数据库连接
cnxn_str = ("Driver={SQL Server Native Client 10.0};"
            "Server=xx.xxx.xxx.xx,xx;"
            "Database=dbase;"
            "UID=sa;"
            "PWD=pswd;")
cnxn = pyodbc.connect(cnxn_str)

# 配置允许查询的表名白名单，从根源避免表名注入风险
ALLOWED_TABLES = {"table1", "table2", "table3"}

def TRX(tbl_name: str, var_value: Union[str, List[str]]):
    # 清洗表名、做白名单校验
    clean_table_name = tbl_name.strip("[]")
    if clean_table_name not in ALLOWED_TABLES:
        raise ValueError(f"无查询表{tbl_name}的权限")
    
    # 适配单值、多值IN查询的参数与占位符生成
    if isinstance(var_value, str):
        query_params = (var_value,)
        placeholders = "?"
    else:
        query_params = tuple(var_value)
        placeholders = ",".join(["?"] * len(query_params))
    
    # 构造SQL语句，表名做方括号转义避免关键字冲突
    query_sql = f"SELECT TOP 100 * FROM [DB].[dbo].[{clean_table_name}] WHERE var1 IN ({placeholders})"
    # 通过params参数传入动态值，禁止直接拼接到SQL字符串中
    result_df = pd.read_sql_query(query_sql, cnxn, params=query_params)
    return result_df

该实现兼容原有调用方式，同时支持传入多值做IN查询：

# 单值查询
df_single = TRX(tbl_name="[table1]", var_value="a")
# 多值查询
df_multi = TRX(tbl_name="[table1]", var_value=["a", "b", "c"])

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