适用于 SQL Server 的 Django + ODBC 驱动程序 17：原始查询无法处理列表或元组参数

Question

代码

我编写了一个函数来包装 Django 对原始 SQL 查询的标准处理。

def run_mssql_query(query, connection="prod", args=None):
    
    """Method will run query on MSSQL database

    Args:
            query (str): MSSQL Query
            connection (str): Which connection to use. Defaults to "prod".

    Returns:
            dict: Results of query
    """
    if connection == "omron":
        results = {}

        conn = get_oracle_connection()

        with conn.cursor() as cursor:
            try:
                if args == None:
                    args = dict()
                    
                cursor.execute(query, args)
                cursor.rowfactory = lambda *args: dict(zip([d[0] for d in cursor.description], args))
                results = cursor.fetchall()
            except:
                results = {}

        return results

    with connections[connection].cursor() as cursor:
        if args == None:
            args = list()
        
        cursor.execute(query, args)  # Replace with your actual query

        try:
            columns = [col[0] for col in cursor.description]
            results = [dict(zip(columns, row)) for row in cursor.fetchall()]
        except:
            results = {}

    return results

执行看起来像这样：

data = run_mssql_query(f"""SELECT *
                           FROM EVENTS.dbo.MES_BARCODES
                           WHERE BARCODE = %s""", args=["123"])

问题

当我使用 string、int 等时...，它就像一个魅力，但是 当我尝试将列表传递给我的参数的参数时，如下所示（例如）：

barcodes = ["123", "321"]
data = run_mssql_query(f"""SELECT *
                           FROM EVENTS.dbo.MES_BARCODES
                           WHERE BARCODE IN %s""", args=[list(barcodes)])

它给了我这个错误：

('42000', '[42000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]Column, parameter, or variable #1: Cannot find data type 321.123. (2715) (SQLExecDirectW)')

这是一个回溯：

File \"C:\\python-projects\\fas-mes\\fasmes\\APP__LogsDashboard\\decorators.py\", line 32, in wrapper\n    response = func(*args, **kwargs)\n  File \"C:\\python-projects\\fas-mes\\fasmes\\APP__Constructors\\views.py\", line 74, in get_duplicated_barcodes\n    data = run_mssql_query(f\"\"\"SELECT *\n  File \"C:\\python-projects\\fas-mes\\fasmes\\API__Database\\views_functions.py\", line 90, in run_mssql_query\n    cursor.execute(query, args)  # Replace with your actual query\n  File \"C:\\python-projects\\fas-mes\\venv\\lib\\site-packages\\django\\db\\backends\\utils.py\", line 102, in execute\n    return super().execute(sql, params)\n  File \"C:\\python-projects\\fas-mes\\venv\\lib\\site-packages\\django\\db\\backends\\utils.py\", line 67, in execute\n    return self._execute_with_wrappers(\n  File \"C:\\python-projects\\fas-mes\\venv\\lib\\site-packages\\django\\db\\backends\\utils.py\", line 80, in _execute_with_wrappers\n    return executor(sql, params, many, context)\n  File \"C:\\python-projects\\fas-mes\\venv\\lib\\site-packages\\django\\db\\backends\\utils.py\", line 89, in _execute\n    return self.cursor.execute(sql, params)\n  File \"C:\\python-projects\\fas-mes\\venv\\lib\\site-packages\\django\\db\\utils.py\", line 91, in __exit__\n    raise dj_exc_value.with_traceback(traceback) from exc_value\n  File \"C:\\python-projects\\fas-mes\\venv\\lib\\site-packages\\django\\db\\backends\\utils.py\", line 89, in _execute\n    return self.cursor.execute(sql, params)\n  File \"C:\\python-projects\\fas-mes\\venv\\lib\\site-packages\\mssql\\base.py\", line 677, in execute\n    return self.cursor.execute(sql, params)\n

我的项目的点冻结

asgiref==3.8.1
backports.zoneinfo==0.2.1
certifi==2024.7.4
cffi==1.17.1
charset-normalizer==3.3.2
cryptography==43.0.1
cx-Oracle==8.3.0
Django==4.2.15
django-appconf==1.0.6
django-compressor==4.5.1
django-cors-headers==4.4.0
django-extensions==3.2.3
django-mssql==1.8
django-mssql-backend==2.8.1
django-sass-processor==1.4.1
django-xff==1.4.0
djangorestframework==3.15.2
idna==3.7
libsass==0.23.0
MarkupSafe==2.1.5
mssql-django==1.5
oracledb==2.4.1
packaging==24.1
pillow==10.4.0
pycparser==2.22
pyodbc==5.1.0
pyOpenSSL==24.2.1
python-ipware==3.0.0
pytz==2024.1
rcssmin==1.1.2
requests==2.32.3
rjsmin==1.2.2
sqlparse==0.5.1
typing-extensions==4.12.2
tzdata==2024.1
urllib3==2.2.2
werkzeug==3.0.4

我尝试做什么

1. 尝试将 ANY() 添加到查询中，如下所示：

data = run_mssql_query(f"""SELECT *
                           FROM EVENTS.dbo.MES_BARCODES
                           WHERE BARCODE IN ANY(%s)""", args=[list(barcodes)])

但这对我一点帮助都没有。

2. 我试图查看 django 尝试执行的查询，然后手动执行它。

这是一个获取查询的函数：

print("Executed queries:" )
for i, query in enumerate(connections[connection].queries):
    print(f'----------------- {i} -------------------')
    print(query["sql"])
    print(f'-----------------------------------------')

这是一个输出：

Executed queries:
----------------- 0 -------------------
SELECT CAST(SERVERPROPERTY('ProductVersion') AS varchar)
-----------------------------------------
----------------- 1 -------------------
SELECT SYSDATETIME()
-----------------------------------------
----------------- 2 -------------------
SELECT *
FROM EVENTS.dbo.MES_BARCODES
WHERE BARCODE IN ['123', '321']
-----------------------------------------

因此，收到的查询看起来绝对正确，并且我认为额外的查询转换正在幕后进行

结果

查看了 StackOverflow 上的一堆文章和官方文档后，发现参数中包含 list() 或 tuple() 的查询不起作用。

附注 我不能使用 f-string 或 string.format 来避免 SQL 注入的风险

Answer 1

您不能以这种方式将绑定参数与使用 Django 的

IN

游标的

connections

子句一起使用。

一种解决方案是动态创建占位符：

with connections["mssql"].cursor() as cursor:
    # Build the SQL with placeholders
    ids = [1, 2, 3, 4]
    placeholders = ", ".join(["%s"] * len(ids))
    sql = f"""
        SELECT *
        FROM EVENTS.dbo.MES_BARCODES
        WHERE BARCODE IN ({placeholders})
    """

    # Run the SQL
    cursor.execute(sql, ids)
    rows = cursor.fetchall()
    for row in rows:
        print(row)

祝你好运！