From 7b7182685bbb024e0ccb82b82dc105790f2d98cc Mon Sep 17 00:00:00 2001
From: Daniele Varrazzo <daniele.varrazzo@gmail.com>
Date: Sat, 4 Apr 2020 00:57:21 +1300
Subject: [PATCH] Added wrapper for escape_bytea with no connection

Changed unescape_bytea to be a normal method. The pgconn is now
optional.
---
 psycopg3/pq/_pq_ctypes.py  |  9 ++++++++-
 psycopg3/pq/_pq_ctypes.pyi |  1 +
 psycopg3/pq/pq_ctypes.py   | 22 +++++++++++++---------
 psycopg3/types/text.py     |  6 ++++--
 tests/pq/test_escaping.py  | 13 ++++++++++++-
 5 files changed, 38 insertions(+), 13 deletions(-)

diff --git a/psycopg3/pq/_pq_ctypes.py b/psycopg3/pq/_pq_ctypes.py
index 135075aac..3dc486452 100644
--- a/psycopg3/pq/_pq_ctypes.py
+++ b/psycopg3/pq/_pq_ctypes.py
@@ -371,7 +371,14 @@ PQescapeByteaConn.argtypes = [
 ]
 PQescapeByteaConn.restype = POINTER(c_ubyte)
 
-# PQescapeBytea: deprecated
+PQescapeBytea = pq.PQescapeBytea
+PQescapeBytea.argtypes = [
+    POINTER(c_char),  # actually POINTER(c_ubyte) but this is easier
+    c_size_t,
+    POINTER(c_size_t),
+]
+PQescapeBytea.restype = POINTER(c_ubyte)
+
 
 PQunescapeBytea = pq.PQunescapeBytea
 PQunescapeBytea.argtypes = [
diff --git a/psycopg3/pq/_pq_ctypes.pyi b/psycopg3/pq/_pq_ctypes.pyi
index 9408d7f7e..340e63205 100644
--- a/psycopg3/pq/_pq_ctypes.pyi
+++ b/psycopg3/pq/_pq_ctypes.pyi
@@ -103,6 +103,7 @@ def PQcmdStatus(arg1: Optional[PGresult_struct]) -> bytes: ...
 def PQcmdTuples(arg1: Optional[PGresult_struct]) -> bytes: ...
 def PQoidValue(arg1: Optional[PGresult_struct]) -> int: ...
 def PQescapeByteaConn(arg1: Optional[PGconn_struct], arg2: bytes, arg3: int, arg4: pointer[c_ulong]) -> pointer[c_ubyte]: ...
+def PQescapeBytea(arg1: bytes, arg2: int, arg3: pointer[c_ulong]) -> pointer[c_ubyte]: ...
 def PQunescapeBytea(arg1: bytes, arg2: pointer[c_ulong]) -> pointer[c_ubyte]: ...
 def PQsendQuery(arg1: Optional[PGconn_struct], arg2: bytes) -> int: ...
 def PQsendQueryParams(arg1: Optional[PGconn_struct], arg2: bytes, arg3: int, arg4: pointer[c_uint], arg5: pointer[c_char_p], arg6: pointer[c_int], arg7: pointer[c_int], arg8: int) -> int: ...
diff --git a/psycopg3/pq/pq_ctypes.py b/psycopg3/pq/pq_ctypes.py
index 626275ed4..e84223dcb 100644
--- a/psycopg3/pq/pq_ctypes.py
+++ b/psycopg3/pq/pq_ctypes.py
@@ -537,17 +537,22 @@ class Conninfo:
 
 
 class Escaping:
-    def __init__(self, conn: PGconn):
+    def __init__(self, conn: Optional[PGconn] = None):
         self.conn = conn
 
     def escape_bytea(self, data: bytes) -> bytes:
         len_out = c_size_t()
-        out = impl.PQescapeByteaConn(
-            self.conn.pgconn_ptr,
-            data,
-            len(data),
-            pointer(t_cast(c_ulong, len_out)),
-        )
+        if self.conn is not None:
+            out = impl.PQescapeByteaConn(
+                self.conn.pgconn_ptr,
+                data,
+                len(data),
+                pointer(t_cast(c_ulong, len_out)),
+            )
+        else:
+            out = impl.PQescapeBytea(
+                data, len(data), pointer(t_cast(c_ulong, len_out)),
+            )
         if not out:
             raise MemoryError(
                 f"couldn't allocate for escape_bytea of {len(data)} bytes"
@@ -557,8 +562,7 @@ class Escaping:
         impl.PQfreemem(out)
         return rv
 
-    @classmethod
-    def unescape_bytea(cls, data: bytes) -> bytes:
+    def unescape_bytea(self, data: bytes) -> bytes:
         len_out = c_size_t()
         out = impl.PQunescapeBytea(data, pointer(t_cast(c_ulong, len_out)))
         if not out:
diff --git a/psycopg3/types/text.py b/psycopg3/types/text.py
index 7ad587880..5533fe80f 100644
--- a/psycopg3/types/text.py
+++ b/psycopg3/types/text.py
@@ -70,7 +70,9 @@ class StringCaster(TypeCaster):
 class BytesAdapter(Adapter):
     def __init__(self, cls: type, conn: BaseConnection):
         super().__init__(cls, conn)
-        self.esc = Escaping(self.conn.pgconn)
+        self.esc = Escaping(
+            self.conn.pgconn if self.conn is not None else None
+        )
 
     def adapt(self, obj: bytes) -> Tuple[bytes, int]:
         return self.esc.escape_bytea(obj), BYTEA_OID
@@ -84,7 +86,7 @@ def adapt_bytes(b: bytes) -> Tuple[bytes, int]:
 @TypeCaster.text(builtins["bytea"].oid)
 @ArrayCaster.text(builtins["bytea"].array_oid)
 def cast_bytea(data: bytes) -> bytes:
-    return Escaping.unescape_bytea(data)
+    return Escaping().unescape_bytea(data)
 
 
 @TypeCaster.binary(builtins["bytea"].oid)
diff --git a/tests/pq/test_escaping.py b/tests/pq/test_escaping.py
index 67495aa2b..01ec91722 100644
--- a/tests/pq/test_escaping.py
+++ b/tests/pq/test_escaping.py
@@ -10,6 +10,17 @@ def test_escape_bytea(pq, pgconn, data):
     assert rv == exp
 
 
+def test_escape_noconn(pq, pgconn):
+    data = bytes(range(256))
+    esc = pq.Escaping()
+    escdata = esc.escape_bytea(data)
+    res = pgconn.exec_params(
+        b"select '%s'::bytea" % escdata, [], result_format=1
+    )
+    assert res.status == pq.ExecStatus.TUPLES_OK
+    assert res.get_value(0, 0) == data
+
+
 def test_escape_1char(pq, pgconn):
     esc = pq.Escaping(pgconn)
     for c in range(256):
@@ -23,5 +34,5 @@ def test_escape_1char(pq, pgconn):
 )
 def test_unescape_bytea(pq, pgconn, data):
     enc = br"\x" + b"".join(b"%02x" % c for c in data)
-    rv = pq.Escaping.unescape_bytea(enc)
+    rv = pq.Escaping(pgconn).unescape_bytea(enc)
     assert rv == data
-- 
2.47.3