Refs #20010 -- Unified DatabaseOperations.last_executed_query() on Oracle with other db backends.

Thanks Simon Charette for the review.
This commit is contained in:
Mariusz Felisiak 2019-04-05 21:35:51 +02:00 committed by GitHub
parent 6b4e57d79f
commit 79065b55a7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 42 additions and 5 deletions

View File

@ -8,7 +8,7 @@ from django.db.backends.base.operations import BaseDatabaseOperations
from django.db.backends.utils import strip_quotes, truncate_name
from django.db.utils import DatabaseError
from django.utils import timezone
from django.utils.encoding import force_bytes
from django.utils.encoding import force_bytes, force_str
from django.utils.functional import cached_property
from .base import Database
@ -258,9 +258,16 @@ END;
# https://cx-oracle.readthedocs.io/en/latest/cursor.html#Cursor.statement
# The DB API definition does not define this attribute.
statement = cursor.statement
# Unlike Psycopg's `query` and MySQLdb`'s `_executed`, CxOracle's
# `statement` doesn't contain the query parameters. refs #20010.
return super().last_executed_query(cursor, statement, params)
# Unlike Psycopg's `query` and MySQLdb`'s `_executed`, cx_Oracle's
# `statement` doesn't contain the query parameters. Substitute
# parameters manually.
if isinstance(params, (tuple, list)):
for i, param in enumerate(params):
statement = statement.replace(':arg%d' % i, force_str(param, errors='replace'))
elif isinstance(params, dict):
for key, param in params.items():
statement = statement.replace(':%s' % key, force_str(param, errors='replace'))
return statement
def last_insert_id(self, cursor, table_name, pk_name):
sq_name = self._get_sequence_name(cursor, strip_quotes(table_name), pk_name)

View File

@ -51,7 +51,7 @@ class DateQuotingTest(TestCase):
@override_settings(DEBUG=True)
class LastExecutedQueryTest(TestCase):
def test_last_executed_query(self):
def test_last_executed_query_without_previous_query(self):
"""
last_executed_query should not raise an exception even if no previous
query has been run.
@ -73,6 +73,36 @@ class LastExecutedQueryTest(TestCase):
last_sql = cursor.db.ops.last_executed_query(cursor, sql, params)
self.assertIsInstance(last_sql, str)
def test_last_executed_query(self):
# last_executed_query() interpolate all parameters, in most cases it is
# not equal to QuerySet.query.
for qs in (
Article.objects.filter(pk=1),
Article.objects.filter(pk__in=(1, 2), reporter__pk=3),
):
sql, params = qs.query.sql_with_params()
cursor = qs.query.get_compiler(DEFAULT_DB_ALIAS).execute_sql(CURSOR)
self.assertEqual(
cursor.db.ops.last_executed_query(cursor, sql, params),
str(qs.query),
)
@skipUnlessDBFeature('supports_paramstyle_pyformat')
def test_last_executed_query_dict(self):
square_opts = Square._meta
sql = 'INSERT INTO %s (%s, %s) VALUES (%%(root)s, %%(square)s)' % (
connection.introspection.identifier_converter(square_opts.db_table),
connection.ops.quote_name(square_opts.get_field('root').column),
connection.ops.quote_name(square_opts.get_field('square').column),
)
with connection.cursor() as cursor:
params = {'root': 2, 'square': 4}
cursor.execute(sql, params)
self.assertEqual(
cursor.db.ops.last_executed_query(cursor, sql, params),
sql % params,
)
class ParameterHandlingTest(TestCase):