Fixed #23546 -- Added kwargs support for CursorWrapper.callproc() on Oracle.

Thanks Shai Berger, Tim Graham and Aymeric Augustin for reviews and
Renbi Yu for the initial patch.
This commit is contained in:
Mariusz Felisiak 2017-08-12 21:06:49 +02:00 committed by GitHub
parent 47ccefeada
commit 489421b015
6 changed files with 46 additions and 9 deletions

View File

@ -240,6 +240,9 @@ class BaseDatabaseFeatures:
create_test_procedure_without_params_sql = None
create_test_procedure_with_int_param_sql = None
# Does the backend support keyword parameters for cursor.callproc()?
supports_callproc_kwargs = False
def __init__(self, connection):
self.connection = connection

View File

@ -54,3 +54,4 @@ class DatabaseFeatures(BaseDatabaseFeatures):
V_I := P_I;
END;
"""
supports_callproc_kwargs = True

View File

@ -6,6 +6,7 @@ import re
from time import time
from django.conf import settings
from django.db.utils import NotSupportedError
from django.utils.encoding import force_bytes
from django.utils.timezone import utc
@ -45,13 +46,23 @@ class CursorWrapper:
# The following methods cannot be implemented in __getattr__, because the
# code must run when the method is invoked, not just when it is accessed.
def callproc(self, procname, params=None):
def callproc(self, procname, params=None, kparams=None):
# Keyword parameters for callproc aren't supported in PEP 249, but the
# database driver may support them (e.g. cx_Oracle).
if kparams is not None and not self.db.features.supports_callproc_kwargs:
raise NotSupportedError(
'Keyword parameters for callproc are not supported on this '
'database backend.'
)
self.db.validate_no_broken_transaction()
with self.db.wrap_database_errors:
if params is None:
if params is None and kparams is None:
return self.cursor.callproc(procname)
else:
elif kparams is None:
return self.cursor.callproc(procname, params)
else:
params = params or ()
return self.cursor.callproc(procname, params, kparams)
def execute(self, sql, params=None):
self.db.validate_no_broken_transaction()

View File

@ -269,6 +269,10 @@ Models
* The new ``field_name`` parameter of :meth:`.QuerySet.in_bulk` allows fetching
results based on any unique model field.
* :meth:`.CursorWrapper.callproc()` now takes an optional dictionary of keyword
parameters, if the backend supports this feature. Of Django's built-in
backends, only Oracle supports it.
Requests and Responses
~~~~~~~~~~~~~~~~~~~~~~

View File

@ -350,10 +350,12 @@ is equivalent to::
Calling stored procedures
~~~~~~~~~~~~~~~~~~~~~~~~~
.. method:: CursorWrapper.callproc(procname, params=None)
.. method:: CursorWrapper.callproc(procname, params=None, kparams=None)
Calls a database stored procedure with the given name and optional sequence
of input parameters.
Calls a database stored procedure with the given name. A sequence
(``params``) or dictionary (``kparams``) of input parameters may be
provided. Most databases don't support ``kparams``. Of Django's built-in
backends, only Oracle supports it.
For example, given this stored procedure in an Oracle database:
@ -372,3 +374,7 @@ Calling stored procedures
with connection.cursor() as cursor:
cursor.callproc('test_procedure', [1, 'test'])
.. versionchanged:: 2.0
The ``kparams`` argument was added.

View File

@ -3,8 +3,9 @@ from decimal import Decimal, Rounded
from django.db import connection
from django.db.backends.utils import format_number, truncate_name
from django.db.utils import NotSupportedError
from django.test import (
SimpleTestCase, TransactionTestCase, skipUnlessDBFeature,
SimpleTestCase, TransactionTestCase, skipIfDBFeature, skipUnlessDBFeature,
)
@ -53,13 +54,13 @@ class TestUtils(SimpleTestCase):
class CursorWrapperTests(TransactionTestCase):
available_apps = []
def _test_procedure(self, procedure_sql, params, param_types):
def _test_procedure(self, procedure_sql, params, param_types, kparams=None):
with connection.cursor() as cursor:
cursor.execute(procedure_sql)
# Use a new cursor because in MySQL a procedure can't be used in the
# same cursor in which it was created.
with connection.cursor() as cursor:
cursor.callproc('test_procedure', params)
cursor.callproc('test_procedure', params, kparams)
with connection.schema_editor() as editor:
editor.remove_procedure('test_procedure', param_types)
@ -70,3 +71,14 @@ class CursorWrapperTests(TransactionTestCase):
@skipUnlessDBFeature('create_test_procedure_with_int_param_sql')
def test_callproc_with_int_params(self):
self._test_procedure(connection.features.create_test_procedure_with_int_param_sql, [1], ['INTEGER'])
@skipUnlessDBFeature('create_test_procedure_with_int_param_sql', 'supports_callproc_kwargs')
def test_callproc_kparams(self):
self._test_procedure(connection.features.create_test_procedure_with_int_param_sql, [], ['INTEGER'], {'P_I': 1})
@skipIfDBFeature('supports_callproc_kwargs')
def test_unsupported_callproc_kparams_raises_error(self):
msg = 'Keyword parameters for callproc are not supported on this database backend.'
with self.assertRaisesMessage(NotSupportedError, msg):
with connection.cursor() as cursor:
cursor.callproc('test_procedure', [], {'P_I': 1})