diff --git a/MySQLdb/cursors.py b/MySQLdb/cursors.py index f8a48640..fdf52c0b 100644 --- a/MySQLdb/cursors.py +++ b/MySQLdb/cursors.py @@ -182,6 +182,15 @@ def execute(self, query, args=None): """ while self.nextset(): pass + + mogrified_query = self._mogrify(query, args) + + assert isinstance(mogrified_query, (bytes, bytearray)) + res = self._query(mogrified_query) + return res + + def _mogrify(self, query, args=None): + """Return query after binding args.""" db = self._get_db() if isinstance(query, str): @@ -202,9 +211,21 @@ def execute(self, query, args=None): except TypeError as m: raise ProgrammingError(str(m)) - assert isinstance(query, (bytes, bytearray)) - res = self._query(query) - return res + return query + + def mogrify(self, query, args=None): + """Return query after binding args. + + query -- string, query to mogrify + args -- optional sequence or mapping, parameters to use with query. + + Note: If args is a sequence, then %s must be used as the + parameter placeholder in the query. If a mapping is used, + %(key)s must be used as the placeholder. + + Returns string representing query that would be executed by the server + """ + return self._mogrify(query, args).decode(self._get_db().encoding) def executemany(self, query, args): # type: (str, list) -> int diff --git a/tests/test_cursor.py b/tests/test_cursor.py index 80e21888..c681b63b 100644 --- a/tests/test_cursor.py +++ b/tests/test_cursor.py @@ -150,3 +150,39 @@ def test_dictcursor(): names2 = sorted(rows[1]) for a, b in zip(names1, names2): assert a is b + + +def test_mogrify_without_args(): + conn = connect() + cursor = conn.cursor() + + query = "SELECT VERSION()" + mogrified_query = cursor.mogrify(query) + cursor.execute(query) + + assert mogrified_query == query + assert mogrified_query == cursor._executed.decode() + + +def test_mogrify_with_tuple_args(): + conn = connect() + cursor = conn.cursor() + + query_with_args = "SELECT %s, %s", (1, 2) + mogrified_query = cursor.mogrify(*query_with_args) + cursor.execute(*query_with_args) + + assert mogrified_query == "SELECT 1, 2" + assert mogrified_query == cursor._executed.decode() + + +def test_mogrify_with_dict_args(): + conn = connect() + cursor = conn.cursor() + + query_with_args = "SELECT %(a)s, %(b)s", {"a": 1, "b": 2} + mogrified_query = cursor.mogrify(*query_with_args) + cursor.execute(*query_with_args) + + assert mogrified_query == "SELECT 1, 2" + assert mogrified_query == cursor._executed.decode()