Skip to content

Commit

Permalink
Add Cursor.mogrify(). (#477)
Browse files Browse the repository at this point in the history
Implements #476

Co-authored-by: Inada Naoki <songofacandy@gmail.com>
  • Loading branch information
Steve Teahan and methane authored May 15, 2023
1 parent 89c1e0f commit 9953e50
Show file tree
Hide file tree
Showing 2 changed files with 60 additions and 3 deletions.
27 changes: 24 additions & 3 deletions MySQLdb/cursors.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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
Expand Down
36 changes: 36 additions & 0 deletions tests/test_cursor.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

0 comments on commit 9953e50

Please sign in to comment.