From 2e7117ab1b584e678380c70625ad1331cea551d0 Mon Sep 17 00:00:00 2001 From: Federico Caselli Date: Tue, 2 Aug 2022 12:27:57 +0200 Subject: 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 --- lib/sqlalchemy/dialects/postgresql/json.py | 47 ++++++++++++++++++++---------- 1 file changed, 31 insertions(+), 16 deletions(-) (limited to 'lib/sqlalchemy/dialects/postgresql/json.py') 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): -- cgit v1.2.1