Django+ODBC Driver 17 for SQL Server原生查询无法处理列表参数求助

我封装了一个处理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"])

单个参数（字符串、整数等）传入时函数运行正常，但尝试传入列表作为IN条件参数时出错：

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

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

• 在查询中添加ANY()：

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

无效果。

• 查看Django实际执行的SQL并手动执行：
  查看SQL的代码：

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']
-----------------------------------------

生成的SQL看起来完全正常，但推测底层存在额外的查询转换。

查阅大量StackOverflow文章和官方文档后，仍无法实现包含列表/元组参数的查询正常运行。
注：为避免SQL注入风险，不能使用f-string或string.format拼接SQL。

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