feat: add ClientCursor.mogrify()

diff --git a/psycopg/psycopg/client_cursor.py b/psycopg/psycopg/client_cursor.py
index 177fb78404964a07cd332ed7a817901dda83c5c2..851b1cec6830bef2214c04f7b6099980b8db9793 100644 (file)
--- a/psycopg/psycopg/client_cursor.py
+++ b/psycopg/psycopg/client_cursor.py
@@ -9,6 +9,7 @@ from functools import partial
 
 from ._queries import PostgresQuery, PostgresClientQuery
 
+from . import adapt
 from . import errors as e
 from .pq import Format
 from .abc import ConnectionType, Query, Params
@@ -23,6 +24,14 @@ if TYPE_CHECKING:
 
 
 class ClientCursorMixin(BaseCursor[ConnectionType, Row]):
+    def mogrify(self, query: Query, params: Optional[Params] = None) -> str:
+        """
+        Return the query to be executed with parameters merged.
+        """
+        self._tx = adapt.Transformer(self)
+        pgq = self._convert_query(query, params)
+        return pgq.query.decode(self._tx.encoding)
+
     def _execute_send(
         self,
         query: PostgresQuery,

diff --git a/tests/test_client_cursor.py b/tests/test_client_cursor.py
index d5b6193f165c181c5fb2b76ebbcb0413243dd09c..dc5b22edb21ac64730aadf1a65074649b1d9261f 100644 (file)
--- a/tests/test_client_cursor.py
+++ b/tests/test_client_cursor.py
@@ -1,6 +1,7 @@
 import gc
 import pickle
 import weakref
+import datetime as dt
 from typing import List
 
 import pytest
@@ -767,3 +768,24 @@ def test_leak(dsn, faker, fetch, row_factory):
         gc_collect()
         n.append(len(gc.get_objects()))
     assert n[0] == n[1] == n[2], f"objects leaked: {n[1] - n[0]}, {n[2] - n[1]}"
+
+
+def test_mogrify(conn):
+    cur = conn.cursor()
+    q = cur.mogrify("select 'hello'")
+    assert q == "select 'hello'"
+
+    q = cur.mogrify("select %s, %s", [1, dt.date(2020, 1, 1)])
+    assert q == "select 1, '2020-01-01'::date"
+
+    conn.execute("set client_encoding to utf8")
+    q = cur.mogrify("select %(s)s", {"s": "\u20ac"})
+    assert q == "select '\u20ac'"
+
+    conn.execute("set client_encoding to latin9")
+    q = cur.mogrify("select %(s)s", {"s": "\u20ac"})
+    assert q == "select '\u20ac'"
+
+    conn.execute("set client_encoding to latin1")
+    with pytest.raises(UnicodeEncodeError):
+        cur.mogrify("select %(s)s", {"s": "\u20ac"})

diff --git a/tests/test_client_cursor_async.py b/tests/test_client_cursor_async.py
index aa0c8f8ef00ce726894b545d832a792aadcdb9c2..f4d7dc8bb3ab854fbf344f3b7b635d15eb2362d6 100644 (file)
--- a/tests/test_client_cursor_async.py
+++ b/tests/test_client_cursor_async.py
@@ -1,6 +1,7 @@
 import gc
 import pytest
 import weakref
+import datetime as dt
 from typing import List
 
 import psycopg
@@ -640,3 +641,24 @@ async def test_leak(dsn, faker, fetch, row_factory):
         n.append(len(gc.get_objects()))
 
     assert n[0] == n[1] == n[2], f"objects leaked: {n[1] - n[0]}, {n[2] - n[1]}"
+
+
+async def test_mogrify(aconn):
+    cur = aconn.cursor()
+    q = cur.mogrify("select 'hello'")
+    assert q == "select 'hello'"
+
+    q = cur.mogrify("select %s, %s", [1, dt.date(2020, 1, 1)])
+    assert q == "select 1, '2020-01-01'::date"
+
+    await aconn.execute("set client_encoding to utf8")
+    q = cur.mogrify("select %(s)s", {"s": "\u20ac"})
+    assert q == "select '\u20ac'"
+
+    await aconn.execute("set client_encoding to latin9")
+    q = cur.mogrify("select %(s)s", {"s": "\u20ac"})
+    assert q == "select '\u20ac'"
+
+    await aconn.execute("set client_encoding to latin1")
+    with pytest.raises(UnicodeEncodeError):
+        cur.mogrify("select %(s)s", {"s": "\u20ac"})