Fixes(#7561) Add support for postgres.UUID literal_binds compilation

Added string rendering to the :class:`.postgresql.UUID` datatype, so that
stringifying a statement with "literal_binds" that uses this type will
render an appropriate string value for the PostgreSQL backend. Pull request
courtesy José Duarte.

Fixes: #7561
Closes: #7563
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/7563
Pull-request-sha: cf6fe73265342d7884a940c4b3a34c9552113ec3

Change-Id: I4b162bdcdce2293a90683e36da54e4a891a3c684
(cherry picked from commit 17d228f6268515bbf37fdd70a6ee3a62cb9a0b0c)

diff --git a/doc/build/changelog/unreleased_14/7561.rst b/doc/build/changelog/unreleased_14/7561.rst
new file mode 100644 (file)
index 0000000..18ea110
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/7561.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: usecase, postgresql
+    :tickets: 7561
+
+    Added string rendering to the :class:`.postgresql.UUID` datatype, so that
+    stringifying a statement with "literal_binds" that uses this type will
+    render an appropriate string value for the PostgreSQL backend. Pull request
+    courtesy José Duarte.

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 04b79a7338e1942e3ba176624e022caad48d15af..ea31a355266cc220c068b2c02daf66a2deac670e 100644 (file)
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1409,7 +1409,6 @@ from ...types import SMALLINT
 from ...types import TEXT
 from ...types import VARCHAR
 
-
 IDX_USING = re.compile(r"^(?:btree|hash|gist|gin|[\w_]+)$", re.I)
 
 AUTOCOMMIT_REGEXP = re.compile(
@@ -1756,6 +1755,24 @@ class UUID(sqltypes.TypeEngine):
         else:
             return None
 
+    def literal_processor(self, dialect):
+        if self.as_uuid:
+
+            def process(value):
+                if value is not None:
+                    value = "'%s'::UUID" % value
+                return value
+
+            return process
+        else:
+
+            def process(value):
+                if value is not None:
+                    value = "'%s'" % value
+                return value
+
+            return process
+
 
 PGUuid = UUID
 

diff --git a/test/dialect/postgresql/test_types.py b/test/dialect/postgresql/test_types.py
index bbd5cadda12de62997d2ee77911354fddba8fbe9..e5b9d48676c593ca88672d16de73e1641806392a 100644 (file)
--- a/test/dialect/postgresql/test_types.py
+++ b/test/dialect/postgresql/test_types.py
@@ -47,6 +47,7 @@ from sqlalchemy.dialects.postgresql import TSTZRANGE
 from sqlalchemy.exc import CompileError
 from sqlalchemy.orm import declarative_base
 from sqlalchemy.orm import Session
+from sqlalchemy.sql import bindparam
 from sqlalchemy.sql import operators
 from sqlalchemy.sql import sqltypes
 from sqlalchemy.sql.type_api import Variant
@@ -922,11 +923,11 @@ class NumericInterpretationTest(fixtures.TestBase):
 
     def test_numeric_codes(self):
         from sqlalchemy.dialects.postgresql import (
+            base,
             pg8000,
             pygresql,
             psycopg2,
             psycopg2cffi,
-            base,
         )
 
         dialects = (
@@ -1415,9 +1416,11 @@ class ArrayTest(AssertsCompiledSQL, fixtures.TestBase):
         argnames="with_enum, using_aggregate_order_by",
     )
     def test_array_agg_specific(self, with_enum, using_aggregate_order_by):
-        from sqlalchemy.dialects.postgresql import aggregate_order_by
-        from sqlalchemy.dialects.postgresql import array_agg
-        from sqlalchemy.dialects.postgresql import ENUM
+        from sqlalchemy.dialects.postgresql import (
+            ENUM,
+            aggregate_order_by,
+            array_agg,
+        )
 
         element_type = ENUM if with_enum else Integer
         expr = (
@@ -2790,6 +2793,33 @@ class UUIDTest(fixtures.TestBase):
     def test_uuid_array(self, datatype, value1, value2, connection):
         self.test_round_trip(datatype, value1, value2, connection)
 
+    @testing.combinations(
+        (
+            "not_as_uuid",
+            postgresql.UUID(as_uuid=False),
+            str(uuid.uuid4()),
+        ),
+        (
+            "as_uuid",
+            postgresql.UUID(as_uuid=True),
+            uuid.uuid4(),
+        ),
+        id_="iaa",
+        argnames="datatype, value1",
+    )
+    def test_uuid_literal(self, datatype, value1, connection):
+        v1 = connection.execute(
+            select(
+                bindparam(
+                    "key",
+                    value=value1,
+                    literal_execute=True,
+                    type_=datatype,
+                )
+            ),
+        )
+        eq_(v1.fetchone()[0], value1)
+
 
 class HStoreTest(AssertsCompiledSQL, fixtures.TestBase):
     __dialect__ = "postgresql"