Refs #14091 -- Fixed connection.queries on SQLite.

This commit is contained in:
Aymeric Augustin 2015-09-13 09:30:35 +02:00
parent fc8a6a9b00
commit 4f6a7663bc
6 changed files with 60 additions and 51 deletions

View File

@ -103,6 +103,39 @@ class DatabaseOperations(BaseDatabaseOperations):
def pk_default_value(self):
return "NULL"
def _quote_params_for_last_executed_query(self, params):
"""
Only for last_executed_query! Don't use this to execute SQL queries!
"""
sql = 'SELECT ' + ', '.join(['QUOTE(?)'] * len(params))
# Bypass Django's wrappers and use the underlying sqlite3 connection
# to avoid logging this query - it would trigger infinite recursion.
cursor = self.connection.connection.cursor()
# Native sqlite3 cursors cannot be used as context managers.
try:
return cursor.execute(sql, params).fetchone()
finally:
cursor.close()
def last_executed_query(self, cursor, sql, params):
# Python substitutes parameters in Modules/_sqlite/cursor.c with:
# pysqlite_statement_bind_parameters(self->statement, parameters, allow_8bit_chars);
# Unfortunately there is no way to reach self->statement from Python,
# so we quote and substitute parameters manually.
if params:
if isinstance(params, (list, tuple)):
params = self._quote_params_for_last_executed_query(params)
else:
keys = params.keys()
values = tuple(params.values())
values = self._quote_params_for_last_executed_query(values)
params = dict(zip(keys, values))
return sql % params
# For consistency with SQLiteCursorWrapper.execute(), just return sql
# when there are no parameters. See #13648 and #17158.
else:
return sql
def quote_name(self, name):
if name.startswith('"') and name.endswith('"'):
return name # Quoting once is enough.

View File

@ -23,8 +23,6 @@ the following::
``connection.queries`` includes all SQL statements -- INSERTs, UPDATES,
SELECTs, etc. Each time your app hits the database, the query will be recorded.
Note that the SQL recorded here may be :ref:`incorrectly quoted under SQLite
<sqlite-connection-queries>`.
If you are using :doc:`multiple databases</topics/db/multi-db>`, you can use the
same interface on each member of the ``connections`` dictionary::

View File

@ -704,16 +704,6 @@ can use the "pyformat" parameter style, where placeholders in the query
are given as ``'%(name)s'`` and the parameters are passed as a dictionary
rather than a list. SQLite does not support this.
.. _sqlite-connection-queries:
Parameters not quoted in ``connection.queries``
-----------------------------------------------
``sqlite3`` does not provide a way to retrieve the SQL after quoting and
substituting the parameters. Instead, the SQL in ``connection.queries`` is
rebuilt with a simple string interpolation. It may be incorrect. Make sure
you add quotes where necessary before copying a query into an SQLite shell.
.. _oracle-notes:
Oracle notes

View File

@ -510,6 +510,8 @@ Models
* Added support for referencing annotations in ``QuerySet.distinct()``.
* ``connection.queries`` shows queries with substituted parameters on SQLite.
CSRF
^^^^

View File

@ -26,7 +26,6 @@ from django.test import (
SimpleTestCase, TestCase, TransactionTestCase, mock, override_settings,
skipIfDBFeature, skipUnlessDBFeature,
)
from django.test.utils import str_prefix
from django.utils import six
from django.utils.six.moves import range
@ -388,8 +387,19 @@ class LastExecutedQueryTest(TestCase):
# This shouldn't raise an exception
query = "SELECT strftime('%Y', 'now');"
connection.cursor().execute(query)
self.assertEqual(connection.queries[-1]['sql'],
str_prefix("QUERY = %(_)s\"SELECT strftime('%%Y', 'now');\" - PARAMS = ()"))
self.assertEqual(connection.queries[-1]['sql'], query)
@unittest.skipUnless(connection.vendor == 'sqlite',
"This test is specific to SQLite.")
def test_parameter_quoting_on_sqlite(self):
# The implementation of last_executed_queries isn't optimal. It's
# worth testing that parameters are quoted. See #14091.
query = "SELECT %s"
params = ["\"'\\"]
connection.cursor().execute(query, params)
# Note that the single quote is repeated
substituted = "SELECT '\"''\\'"
self.assertEqual(connection.queries[-1]['sql'], substituted)
class ParameterHandlingTest(TestCase):

View File

@ -61,28 +61,14 @@ class TestDebugSQL(unittest.TestCase):
for output in self.verbose_expected_outputs:
self.assertIn(output, full_output)
if six.PY3:
expected_outputs = [
('''QUERY = 'SELECT COUNT(*) AS "__count" '''
'''FROM "test_runner_person" WHERE '''
'''"test_runner_person"."first_name" = %s' '''
'''- PARAMS = ('error',);'''),
('''QUERY = 'SELECT COUNT(*) AS "__count" '''
'''FROM "test_runner_person" WHERE '''
'''"test_runner_person"."first_name" = %s' '''
'''- PARAMS = ('fail',);'''),
]
else:
expected_outputs = [
('''QUERY = u'SELECT COUNT(*) AS "__count" '''
'''FROM "test_runner_person" WHERE '''
'''"test_runner_person"."first_name" = %s' '''
'''- PARAMS = (u'error',);'''),
('''QUERY = u'SELECT COUNT(*) AS "__count" '''
'''FROM "test_runner_person" WHERE '''
'''"test_runner_person"."first_name" = %s' '''
'''- PARAMS = (u'fail',);'''),
]
expected_outputs = [
('''SELECT COUNT(*) AS "__count" '''
'''FROM "test_runner_person" WHERE '''
'''"test_runner_person"."first_name" = 'error';'''),
('''SELECT COUNT(*) AS "__count" '''
'''FROM "test_runner_person" WHERE '''
'''"test_runner_person"."first_name" = 'fail';'''),
]
verbose_expected_outputs = [
# Output format changed in Python 3.5+
@ -91,18 +77,8 @@ class TestDebugSQL(unittest.TestCase):
'runTest (test_runner.test_debug_sql.{}ErrorTest) ... ERROR',
'runTest (test_runner.test_debug_sql.{}PassingTest) ... ok',
]
] + [
('''SELECT COUNT(*) AS "__count" '''
'''FROM "test_runner_person" WHERE '''
'''"test_runner_person"."first_name" = 'pass';'''),
]
if six.PY3:
verbose_expected_outputs += [
('''QUERY = 'SELECT COUNT(*) AS "__count" '''
'''FROM "test_runner_person" WHERE '''
'''"test_runner_person"."first_name" = %s' '''
'''- PARAMS = ('pass',);'''),
]
else:
verbose_expected_outputs += [
('''QUERY = u'SELECT COUNT(*) AS "__count" '''
'''FROM "test_runner_person" WHERE '''
'''"test_runner_person"."first_name" = %s' '''
'''- PARAMS = (u'pass',);'''),
]