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

4 files changed, 49 insertions, 21 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/__init__.py b/lib/sqlalchemy/dialects/postgresql/__init__.py
index 104077a17..8dbee1f7f 100644
--- a/lib/sqlalchemy/dialects/postgresql/__init__.py
+++ b/lib/sqlalchemy/dialects/postgresql/__init__.py
@@ -41,6 +41,7 @@ from .hstore import HSTORE
 from .hstore import hstore
 from .json import JSON
 from .json import JSONB
+from .json import JSONPATH
 from .named_types import CreateDomainType
 from .named_types import CreateEnumType
 from .named_types import DropDomainType
@@ -128,6 +129,7 @@ __all__ = (
     "TSTZMULTIRANGE",
     "JSON",
     "JSONB",
+    "JSONPATH",
     "Any",
     "All",
     "DropEnumType",
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index 38f8fddee..6888959f0 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -122,7 +122,6 @@ client using this setting passed to :func:`_asyncio.create_async_engine`::
 from __future__ import annotations
 
 import collections
-import collections.abc as collections_abc
 import decimal
 import json as _py_json
 import re
@@ -231,9 +230,15 @@ class AsyncpgJSONStrIndexType(sqltypes.JSON.JSONStrIndexType):
 class AsyncpgJSONPathType(json.JSONPathType):
     def bind_processor(self, dialect):
         def process(value):
-            assert isinstance(value, collections_abc.Sequence)
-            tokens = [str(elem) for elem in value]
-            return 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:
+                tokens = [str(elem) for elem in value]
+                return tokens
+            else:
+                return []
 
         return process
 
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 35a09cfa7..75fa3d7c7 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1512,7 +1512,7 @@ colspecs = {
     sqltypes.ARRAY: _array.ARRAY,
     sqltypes.Interval: INTERVAL,
     sqltypes.Enum: ENUM,
-    sqltypes.JSON.JSONPathType: _json.JSONPathType,
+    sqltypes.JSON.JSONPathType: _json.JSONPATH,
     sqltypes.JSON: _json.JSON,
     UUID: PGUuid,
 }
@@ -2503,6 +2503,12 @@ class PGTypeCompiler(compiler.GenericTypeCompiler):
             count=1,
         )
 
+    def visit_json_path(self, type_, **kw):
+        return self.visit_JSONPATH(type_, **kw)
+
+    def visit_JSONPATH(self, type_, **kw):
+        return "JSONPATH"
+
 
 class PGIdentifierPreparer(compiler.IdentifierPreparer):
 
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):