diff --git a/django/db/backends/oracle/operations.py b/django/db/backends/oracle/operations.py index 6ac6b29737..c1afb2ed5e 100644 --- a/django/db/backends/oracle/operations.py +++ b/django/db/backends/oracle/operations.py @@ -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) diff --git a/tests/backends/tests.py b/tests/backends/tests.py index 6138a3626c..7d5dbe7a39 100644 --- a/tests/backends/tests.py +++ b/tests/backends/tests.py @@ -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):