Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/json.py: 60%

1# dialects/postgresql/json.py 

2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# 

5# This module is part of SQLAlchemy and is released under 

6# the MIT License: https://www.opensource.org/licenses/mit-license.php 

7 

8from __future__ import annotations 

9 

10from typing import Any 

11from typing import Callable 

12from typing import List 

13from typing import Optional 

14from typing import TYPE_CHECKING 

15from typing import Union 

16 

17from .array import ARRAY 

18from .array import array as _pg_array 

19from .operators import ASTEXT 

20from .operators import CONTAINED_BY 

21from .operators import CONTAINS 

22from .operators import DELETE_PATH 

23from .operators import HAS_ALL 

24from .operators import HAS_ANY 

25from .operators import HAS_KEY 

26from .operators import JSONPATH_ASTEXT 

27from .operators import PATH_EXISTS 

28from .operators import PATH_MATCH 

29from ... import types as sqltypes 

30from ...sql import cast 

31from ...sql._typing import _T 

32 

33if TYPE_CHECKING: 

34 from ...engine.interfaces import Dialect 

35 from ...sql.elements import ColumnElement 

36 from ...sql.type_api import _BindProcessorType 

37 from ...sql.type_api import _LiteralProcessorType 

38 from ...sql.type_api import TypeEngine 

39 

40__all__ = ("JSON", "JSONB") 

41 

42 

43class JSONPathType(sqltypes.JSON.JSONPathType): 

44 def _processor( 

45 self, dialect: Dialect, super_proc: Optional[Callable[[Any], Any]] 

46 ) -> Callable[[Any], Any]: 

47 def process(value: Any) -> Any: 

48 if isinstance(value, str): 

49 # If it's already a string assume that it's in json path 

50 # format. This allows using cast with json paths literals 

51 return value 

52 elif value: 

53 # If it's already a string assume that it's in json path 

54 # format. This allows using cast with json paths literals 

55 value = "{%s}" % (", ".join(map(str, value))) 

56 else: 

57 value = "{}" 

58 if super_proc: 

59 value = super_proc(value) 

60 return value 

61 

62 return process 

63 

64 def bind_processor(self, dialect: Dialect) -> _BindProcessorType[Any]: 

65 return self._processor(dialect, self.string_bind_processor(dialect)) # type: ignore[return-value] # noqa: E501 

66 

67 def literal_processor( 

68 self, dialect: Dialect 

69 ) -> _LiteralProcessorType[Any]: 

70 return self._processor(dialect, self.string_literal_processor(dialect)) # type: ignore[return-value] # noqa: E501 

71 

72 

73class JSONPATH(JSONPathType): 

74 """JSON Path Type. 

75 

76 This is usually required to cast literal values to json path when using 

77 json search like function, such as ``jsonb_path_query_array`` or 

78 ``jsonb_path_exists``:: 

79 

80 stmt = sa.select( 

81 sa.func.jsonb_path_query_array( 

82 table.c.jsonb_col, cast("$.address.id", JSONPATH) 

83 ) 

84 ) 

85 

86 """ 

87 

88 __visit_name__ = "JSONPATH" 

89 

90 

91class JSON(sqltypes.JSON): 

92 """Represent the PostgreSQL JSON type. 

93 

94 :class:`_postgresql.JSON` is used automatically whenever the base 

95 :class:`_types.JSON` datatype is used against a PostgreSQL backend, 

96 however base :class:`_types.JSON` datatype does not provide Python 

97 accessors for PostgreSQL-specific comparison methods such as 

98 :meth:`_postgresql.JSON.Comparator.astext`; additionally, to use 

99 PostgreSQL ``JSONB``, the :class:`_postgresql.JSONB` datatype should 

100 be used explicitly. 

101 

102 .. seealso:: 

103 

104 :class:`_types.JSON` - main documentation for the generic 

105 cross-platform JSON datatype. 

106 

107 The operators provided by the PostgreSQL version of :class:`_types.JSON` 

108 include: 

109 

110 * Index operations (the ``->`` operator):: 

111 

112 data_table.c.data["some key"] 

113 

114 data_table.c.data[5] 

115 

116 * Index operations returning text 

117 (the ``->>`` operator):: 

118 

119 data_table.c.data["some key"].astext == "some value" 

120 

121 Note that equivalent functionality is available via the 

122 :attr:`.JSON.Comparator.as_string` accessor. 

123 

124 * Index operations with CAST 

125 (equivalent to ``CAST(col ->> ['some key'] AS <type>)``):: 

126 

127 data_table.c.data["some key"].astext.cast(Integer) == 5 

128 

129 Note that equivalent functionality is available via the 

130 :attr:`.JSON.Comparator.as_integer` and similar accessors. 

131 

132 * Path index operations (the ``#>`` operator):: 

133 

134 data_table.c.data[("key_1", "key_2", 5, ..., "key_n")] 

135 

136 * Path index operations returning text (the ``#>>`` operator):: 

137 

138 data_table.c.data[ 

139 ("key_1", "key_2", 5, ..., "key_n") 

140 ].astext == "some value" 

141 

142 Index operations return an expression object whose type defaults to 

143 :class:`_types.JSON` by default, 

144 so that further JSON-oriented instructions 

145 may be called upon the result type. 

146 

147 Custom serializers and deserializers are specified at the dialect level, 

148 that is using :func:`_sa.create_engine`. The reason for this is that when 

149 using psycopg2, the DBAPI only allows serializers at the per-cursor 

150 or per-connection level. E.g.:: 

151 

152 engine = create_engine( 

153 "postgresql+psycopg2://scott:tiger@localhost/test", 

154 json_serializer=my_serialize_fn, 

155 json_deserializer=my_deserialize_fn, 

156 ) 

157 

158 When using the psycopg2 dialect, the json_deserializer is registered 

159 against the database using ``psycopg2.extras.register_default_json``. 

160 

161 .. seealso:: 

162 

163 :class:`_types.JSON` - Core level JSON type 

164 

165 :class:`_postgresql.JSONB` 

166 

167 """ # noqa 

168 

169 render_bind_cast = True 

170 astext_type: TypeEngine[str] = sqltypes.Text() 

171 

172 def __init__( 

173 self, 

174 none_as_null: bool = False, 

175 astext_type: Optional[TypeEngine[str]] = None, 

176 ): 

177 """Construct a :class:`_types.JSON` type. 

178 

179 :param none_as_null: if True, persist the value ``None`` as a 

180 SQL NULL value, not the JSON encoding of ``null``. Note that 

181 when this flag is False, the :func:`.null` construct can still 

182 be used to persist a NULL value:: 

183 

184 from sqlalchemy import null 

185 

186 conn.execute(table.insert(), {"data": null()}) 

187 

188 .. seealso:: 

189 

190 :attr:`_types.JSON.NULL` 

191 

192 :param astext_type: the type to use for the 

193 :attr:`.JSON.Comparator.astext` 

194 accessor on indexed attributes. Defaults to :class:`_types.Text`. 

195 

196 """ 

197 super().__init__(none_as_null=none_as_null) 

198 if astext_type is not None: 

199 self.astext_type = astext_type 

200 

201 class Comparator(sqltypes.JSON.Comparator[_T]): 

202 """Define comparison operations for :class:`_types.JSON`.""" 

203 

204 type: JSON 

205 

206 @property 

207 def astext(self) -> ColumnElement[str]: 

208 """On an indexed expression, use the "astext" (e.g. "->>") 

209 conversion when rendered in SQL. 

210 

211 E.g.:: 

212 

213 select(data_table.c.data["some key"].astext) 

214 

215 .. seealso:: 

216 

217 :meth:`_expression.ColumnElement.cast` 

218 

219 """ 

220 if isinstance(self.expr.right.type, sqltypes.JSON.JSONPathType): 

221 return self.expr.left.operate( # type: ignore[no-any-return] 

222 JSONPATH_ASTEXT, 

223 self.expr.right, 

224 result_type=self.type.astext_type, 

225 ) 

226 else: 

227 return self.expr.left.operate( # type: ignore[no-any-return] 

228 ASTEXT, self.expr.right, result_type=self.type.astext_type 

229 ) 

230 

231 comparator_factory = Comparator 

232 

233 

234class JSONB(JSON): 

235 """Represent the PostgreSQL JSONB type. 

236 

237 The :class:`_postgresql.JSONB` type stores arbitrary JSONB format data, 

238 e.g.:: 

239 

240 data_table = Table( 

241 "data_table", 

242 metadata, 

243 Column("id", Integer, primary_key=True), 

244 Column("data", JSONB), 

245 ) 

246 

247 with engine.connect() as conn: 

248 conn.execute( 

249 data_table.insert(), data={"key1": "value1", "key2": "value2"} 

250 ) 

251 

252 The :class:`_postgresql.JSONB` type includes all operations provided by 

253 :class:`_types.JSON`, including the same behaviors for indexing 

254 operations. 

255 It also adds additional operators specific to JSONB, including 

256 :meth:`.JSONB.Comparator.has_key`, :meth:`.JSONB.Comparator.has_all`, 

257 :meth:`.JSONB.Comparator.has_any`, :meth:`.JSONB.Comparator.contains`, 

258 :meth:`.JSONB.Comparator.contained_by`, 

259 :meth:`.JSONB.Comparator.delete_path`, 

260 :meth:`.JSONB.Comparator.path_exists` and 

261 :meth:`.JSONB.Comparator.path_match`. 

262 

263 Like the :class:`_types.JSON` type, the :class:`_postgresql.JSONB` 

264 type does not detect 

265 in-place changes when used with the ORM, unless the 

266 :mod:`sqlalchemy.ext.mutable` extension is used. 

267 

268 Custom serializers and deserializers 

269 are shared with the :class:`_types.JSON` class, 

270 using the ``json_serializer`` 

271 and ``json_deserializer`` keyword arguments. These must be specified 

272 at the dialect level using :func:`_sa.create_engine`. When using 

273 psycopg2, the serializers are associated with the jsonb type using 

274 ``psycopg2.extras.register_default_jsonb`` on a per-connection basis, 

275 in the same way that ``psycopg2.extras.register_default_json`` is used 

276 to register these handlers with the json type. 

277 

278 .. seealso:: 

279 

280 :class:`_types.JSON` 

281 

282 """ 

283 

284 __visit_name__ = "JSONB" 

285 

286 class Comparator(JSON.Comparator[_T]): 

287 """Define comparison operations for :class:`_types.JSON`.""" 

288 

289 type: JSONB 

290 

291 def has_key(self, other: Any) -> ColumnElement[bool]: 

292 """Boolean expression. Test for presence of a key (equivalent of 

293 the ``?`` operator). Note that the key may be a SQLA expression. 

294 """ 

295 return self.operate(HAS_KEY, other, result_type=sqltypes.Boolean) 

296 

297 def has_all(self, other: Any) -> ColumnElement[bool]: 

298 """Boolean expression. Test for presence of all keys in jsonb 

299 (equivalent of the ``?&`` operator) 

300 """ 

301 return self.operate(HAS_ALL, other, result_type=sqltypes.Boolean) 

302 

303 def has_any(self, other: Any) -> ColumnElement[bool]: 

304 """Boolean expression. Test for presence of any key in jsonb 

305 (equivalent of the ``?|`` operator) 

306 """ 

307 return self.operate(HAS_ANY, other, result_type=sqltypes.Boolean) 

308 

309 def contains(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: 

310 """Boolean expression. Test if keys (or array) are a superset 

311 of/contained the keys of the argument jsonb expression 

312 (equivalent of the ``@>`` operator). 

313 

314 kwargs may be ignored by this operator but are required for API 

315 conformance. 

316 """ 

317 return self.operate(CONTAINS, other, result_type=sqltypes.Boolean) 

318 

319 def contained_by(self, other: Any) -> ColumnElement[bool]: 

320 """Boolean expression. Test if keys are a proper subset of the 

321 keys of the argument jsonb expression 

322 (equivalent of the ``<@`` operator). 

323 """ 

324 return self.operate( 

325 CONTAINED_BY, other, result_type=sqltypes.Boolean 

326 ) 

327 

328 def delete_path( 

329 self, array: Union[List[str], _pg_array[str]] 

330 ) -> ColumnElement[JSONB]: 

331 """JSONB expression. Deletes field or array element specified in 

332 the argument array (equivalent of the ``#-`` operator). 

333 

334 The input may be a list of strings that will be coerced to an 

335 ``ARRAY`` or an instance of :meth:`_postgres.array`. 

336 

337 .. versionadded:: 2.0 

338 """ 

339 if not isinstance(array, _pg_array): 

340 array = _pg_array(array) 

341 right_side = cast(array, ARRAY(sqltypes.TEXT)) 

342 return self.operate(DELETE_PATH, right_side, result_type=JSONB) 

343 

344 def path_exists(self, other: Any) -> ColumnElement[bool]: 

345 """Boolean expression. Test for presence of item given by the 

346 argument JSONPath expression (equivalent of the ``@?`` operator). 

347 

348 .. versionadded:: 2.0 

349 """ 

350 return self.operate( 

351 PATH_EXISTS, other, result_type=sqltypes.Boolean 

352 ) 

353 

354 def path_match(self, other: Any) -> ColumnElement[bool]: 

355 """Boolean expression. Test if JSONPath predicate given by the 

356 argument JSONPath expression matches 

357 (equivalent of the ``@@`` operator). 

358 

359 Only the first item of the result is taken into account. 

360 

361 .. versionadded:: 2.0 

362 """ 

363 return self.operate( 

364 PATH_MATCH, other, result_type=sqltypes.Boolean 

365 ) 

366 

367 comparator_factory = Comparator