I wrote a function that wraps Django's standard processing for raw SQL queries.
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
Execution looks something like this:
data = run_mssql_query(f"""SELECT *
FROM EVENTS.dbo.MES_BARCODES
WHERE BARCODE = %s""", args=["123"])
When i'm using string, int, etc..., it's working like a charm, but when i'm tring to pass list to the params of my params, like this (for example):
barcodes = ["123", "321"]
data = run_mssql_query(f"""SELECT *
FROM EVENTS.dbo.MES_BARCODES
WHERE BARCODE IN %s""", args=[list(barcodes)])
It's giving me this error:
('42000', '[42000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]Column, parameter, or variable #1: Cannot find data type 321.123. (2715) (SQLExecDirectW)')
Here's a traceback:
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
data = run_mssql_query(f"""SELECT *
FROM EVENTS.dbo.MES_BARCODES
WHERE BARCODE IN ANY(%s)""", args=[list(barcodes)])
But that's not helped me at all.
Here's a function to get queries:
print("Executed queries:" )
for i, query in enumerate(connections[connection].queries):
print(f'----------------- {i} -------------------')
print(query["sql"])
print(f'-----------------------------------------')
Here's an output:
Executed queries:
----------------- 0 -------------------
SELECT CAST(SERVERPROPERTY('ProductVersion') AS varchar)
-----------------------------------------
----------------- 1 -------------------
SELECT SYSDATETIME()
-----------------------------------------
----------------- 2 -------------------
SELECT *
FROM EVENTS.dbo.MES_BARCODES
WHERE BARCODE IN ['123', '321']
-----------------------------------------
As a result, the received queries look absolutely correct, and I think that additional query transformations are taking place under the hood
After viewing a bunch of articles on StackOverflow and official documentation, it did not work to make a query containing list() or tuple() in arguments work.
P.S. I can't use f-string or string.format in a risk of SQL-injections
You can't use a bound parameter in this fashion with an IN clause using Django's connections cursors.
One solution is to dynamically create the placeholders:
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)
Good luck!