JSONPATH type can be used in casts in PostgreSQL

Introduced the type :class:`_postgresql.JSONPATH` that can be used
in cast expressions. This is required by some PostgreSQL dialects
when using functions such as ``jsonb_path_exists`` or
``jsonb_path_match`` that accept a ``jsonpath`` as input.

Fixes: #8216
Change-Id: I3e7337eab91680cab1604e1f3058854a0a19c5be

1 files changed, 31 insertions, 16 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/json.py b/lib/sqlalchemy/dialects/postgresql/json.py
index 8763a0ca2..a8b03bd48 100644
--- a/lib/sqlalchemy/dialects/postgresql/json.py
+++ b/lib/sqlalchemy/dialects/postgresql/json.py
@@ -6,7 +6,6 @@
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 # mypy: ignore-errors
 
-import collections.abc as collections_abc
 
 from ... import types as sqltypes
 from ...sql import operators
@@ -68,31 +67,47 @@ CONTAINED_BY = operators.custom_op(
 
 
 class JSONPathType(sqltypes.JSON.JSONPathType):
-    def bind_processor(self, dialect):
-        super_proc = self.string_bind_processor(dialect)
-
+    def _processor(self, dialect, super_proc):
         def process(value):
-            assert isinstance(value, collections_abc.Sequence)
-            tokens = [str(elem) for elem in value]
-            value = "{%s}" % (", ".join(tokens))
+            if isinstance(value, str):
+                # If it's already a string assume that it's in json path
+                # format. This allows using cast with json paths literals
+                return value
+            elif value:
+                # If it's already a string assume that it's in json path
+                # format. This allows using cast with json paths literals
+                value = "{%s}" % (", ".join(map(str, value)))
+            else:
+                value = "{}"
             if super_proc:
                 value = super_proc(value)
             return value
 
         return process
 
+    def bind_processor(self, dialect):
+        return self._processor(dialect, self.string_bind_processor(dialect))
+
     def literal_processor(self, dialect):
-        super_proc = self.string_literal_processor(dialect)
+        return self._processor(dialect, self.string_literal_processor(dialect))
 
-        def process(value):
-            assert isinstance(value, collections_abc.Sequence)
-            tokens = [str(elem) for elem in value]
-            value = "{%s}" % (", ".join(tokens))
-            if super_proc:
-                value = super_proc(value)
-            return value
 
-        return process
+class JSONPATH(JSONPathType):
+    """JSON Path Type.
+
+    This is usually required to cast literal values to json path when using
+    json search like function, such as ``jsonb_path_query_array`` or
+    ``jsonb_path_exists``::
+
+        stmt = sa.select(
+            sa.func.jsonb_path_query_array(
+                table.c.jsonb_col, cast("$.address.id", JSONPATH)
+            )
+        )
+
+    """
+
+    __visit_name__ = "JSONPATH"
 
 
 class JSON(sqltypes.JSON):