Added mogrify support to Cursor (PyMySQL#476)

Author: steveteahan

MySQLdb/cursors.py

@@ -182,6 +182,27 @@ 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):
+        """Generates the query to be sent to the server with argument
+        interpolation and proper encoding. This is for internal use, see
+        mogrify() for the external API that returns a string.
+
+        query -- string, query to execute on server
+        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 bytes or bytearray representing the query
+        """
         db = self._get_db()
 
         if isinstance(query, str):
@@ -202,9 +223,22 @@ 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):
+        """Get the query exactly as it would be sent to the database running the
+        execute() method.
+
+        query -- string, query to execute on server
+        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()
 
     def executemany(self, query, args):
         # type: (str, list) -> int

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()