Fixed #36420 -- Used actual SQLite limits in last_executed_query() quoting.
Some checks failed
Docs / spelling (push) Has been cancelled
Docs / blacken-docs (push) Has been cancelled
Docs / lint-docs (push) Has been cancelled
Linters / flake8 (push) Has been cancelled
Linters / isort (push) Has been cancelled
Linters / black (push) Has been cancelled
Tests / Windows, SQLite, Python 3.14 (push) Has been cancelled
Tests / JavaScript tests (push) Has been cancelled

This commit is contained in:
myoungjinGo 2025-07-16 00:43:45 +09:00 committed by Jacob Walls
parent 1c7db70e79
commit c4e07f94eb
2 changed files with 32 additions and 18 deletions

View file

@ -1,5 +1,6 @@
import datetime import datetime
import decimal import decimal
import sqlite3
import uuid import uuid
from functools import lru_cache from functools import lru_cache
from itertools import chain from itertools import chain
@ -143,16 +144,15 @@ class DatabaseOperations(BaseDatabaseOperations):
""" """
Only for last_executed_query! Don't use this to execute SQL queries! Only for last_executed_query! Don't use this to execute SQL queries!
""" """
# This function is limited both by SQLITE_LIMIT_VARIABLE_NUMBER (the connection = self.connection.connection
# number of parameters, default = 999) and SQLITE_MAX_COLUMN (the variable_limit = self.connection.features.max_query_params
# number of return values, default = 2000). Since Python's sqlite3 column_limit = connection.getlimit(sqlite3.SQLITE_LIMIT_COLUMN)
# module doesn't expose the get_limit() C API, assume the default batch_size = min(variable_limit, column_limit)
# limits are in effect and split the work in batches if needed.
BATCH_SIZE = 999 if len(params) > batch_size:
if len(params) > BATCH_SIZE:
results = () results = ()
for index in range(0, len(params), BATCH_SIZE): for index in range(0, len(params), batch_size):
chunk = params[index : index + BATCH_SIZE] chunk = params[index : index + batch_size]
results += self._quote_params_for_last_executed_query(chunk) results += self._quote_params_for_last_executed_query(chunk)
return results return results

View file

@ -1,5 +1,6 @@
import os import os
import re import re
import sqlite3
import tempfile import tempfile
import threading import threading
import unittest import unittest
@ -215,15 +216,28 @@ class LastExecutedQueryTest(TestCase):
substituted = "SELECT '\"''\\'" substituted = "SELECT '\"''\\'"
self.assertEqual(connection.queries[-1]["sql"], substituted) self.assertEqual(connection.queries[-1]["sql"], substituted)
def test_large_number_of_parameters(self): def test_parameter_count_exceeds_variable_or_column_limit(self):
# If SQLITE_MAX_VARIABLE_NUMBER (default = 999) has been changed to be sql = "SELECT MAX(%s)" % ", ".join(["%s"] * 1001)
# greater than SQLITE_MAX_COLUMN (default = 2000), last_executed_query params = list(range(1001))
# can hit the SQLITE_MAX_COLUMN limit (#26063). for label, limit, current_limit in [
with connection.cursor() as cursor: (
sql = "SELECT MAX(%s)" % ", ".join(["%s"] * 2001) "variable",
params = list(range(2001)) sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER,
# This should not raise an exception. connection.features.max_query_params,
cursor.db.ops.last_executed_query(cursor.cursor, sql, params) ),
(
"column",
sqlite3.SQLITE_LIMIT_COLUMN,
connection.connection.getlimit(sqlite3.SQLITE_LIMIT_COLUMN),
),
]:
with self.subTest(limit=label):
connection.connection.setlimit(limit, 1000)
self.addCleanup(connection.connection.setlimit, limit, current_limit)
with connection.cursor() as cursor:
# This should not raise an exception.
cursor.db.ops.last_executed_query(cursor.cursor, sql, params)
connection.connection.setlimit(limit, current_limit)
@unittest.skipUnless(connection.vendor == "sqlite", "SQLite tests") @unittest.skipUnless(connection.vendor == "sqlite", "SQLite tests")