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

1# dialects/postgresql/hstore.py 

2# Copyright (C) 2005-2026 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 

10import re 

11from typing import Any 

12from typing import Optional 

13 

14from .array import ARRAY 

15from .operators import CONTAINED_BY 

16from .operators import CONTAINS 

17from .operators import GETITEM 

18from .operators import HAS_ALL 

19from .operators import HAS_ANY 

20from .operators import HAS_KEY 

21from ... import types as sqltypes 

22from ...sql import functions as sqlfunc 

23from ...types import OperatorClass 

24 

25__all__ = ("HSTORE", "hstore") 

26 

27_HSTORE_VAL = dict[str, str | None] 

28 

29 

30class HSTORE( 

31 sqltypes.Indexable, 

32 sqltypes.Concatenable, 

33 sqltypes.TypeEngine[_HSTORE_VAL], 

34): 

35 """Represent the PostgreSQL HSTORE type. 

36 

37 The :class:`.HSTORE` type stores dictionaries containing strings, e.g.:: 

38 

39 data_table = Table( 

40 "data_table", 

41 metadata, 

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

43 Column("data", HSTORE), 

44 ) 

45 

46 with engine.connect() as conn: 

47 conn.execute( 

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

49 ) 

50 

51 :class:`.HSTORE` provides for a wide range of operations, including: 

52 

53 * Index operations:: 

54 

55 data_table.c.data["some key"] == "some value" 

56 

57 * Containment operations:: 

58 

59 data_table.c.data.has_key("some key") 

60 

61 data_table.c.data.has_all(["one", "two", "three"]) 

62 

63 * Concatenation:: 

64 

65 data_table.c.data + {"k1": "v1"} 

66 

67 For a full list of special methods see 

68 :class:`.HSTORE.comparator_factory`. 

69 

70 .. container:: topic 

71 

72 **Detecting Changes in HSTORE columns when using the ORM** 

73 

74 For usage with the SQLAlchemy ORM, it may be desirable to combine the 

75 usage of :class:`.HSTORE` with :class:`.MutableDict` dictionary now 

76 part of the :mod:`sqlalchemy.ext.mutable` extension. This extension 

77 will allow "in-place" changes to the dictionary, e.g. addition of new 

78 keys or replacement/removal of existing keys to/from the current 

79 dictionary, to produce events which will be detected by the unit of 

80 work:: 

81 

82 from sqlalchemy.ext.mutable import MutableDict 

83 

84 

85 class MyClass(Base): 

86 __tablename__ = "data_table" 

87 

88 id = Column(Integer, primary_key=True) 

89 data = Column(MutableDict.as_mutable(HSTORE)) 

90 

91 

92 my_object = session.query(MyClass).one() 

93 

94 # in-place mutation, requires Mutable extension 

95 # in order for the ORM to detect 

96 my_object.data["some_key"] = "some value" 

97 

98 session.commit() 

99 

100 When the :mod:`sqlalchemy.ext.mutable` extension is not used, the ORM 

101 will not be alerted to any changes to the contents of an existing 

102 dictionary, unless that dictionary value is re-assigned to the 

103 HSTORE-attribute itself, thus generating a change event. 

104 

105 .. seealso:: 

106 

107 :class:`.hstore` - render the PostgreSQL ``hstore()`` function. 

108 

109 

110 """ # noqa: E501 

111 

112 __visit_name__ = "HSTORE" 

113 hashable = False 

114 text_type = sqltypes.Text() 

115 

116 operator_classes = ( 

117 OperatorClass.BASE 

118 | OperatorClass.CONTAINS 

119 | OperatorClass.INDEXABLE 

120 | OperatorClass.CONCATENABLE 

121 ) 

122 

123 def __init__(self, text_type: Optional[Any] = None) -> None: 

124 """Construct a new :class:`.HSTORE`. 

125 

126 :param text_type: the type that should be used for indexed values. 

127 Defaults to :class:`_types.Text`. 

128 

129 """ 

130 if text_type is not None: 

131 self.text_type = text_type 

132 

133 class Comparator( 

134 sqltypes.Indexable.Comparator[_HSTORE_VAL], 

135 sqltypes.Concatenable.Comparator[_HSTORE_VAL], 

136 ): 

137 """Define comparison operations for :class:`.HSTORE`.""" 

138 

139 def has_key(self, other: Any) -> Any: 

140 """Boolean expression. Test for presence of a key. Note that the 

141 key may be a SQLA expression. 

142 """ 

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

144 

145 def has_all(self, other: Any) -> Any: 

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

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

148 

149 def has_any(self, other: Any) -> Any: 

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

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

152 

153 def contains(self, other: Any, **kwargs: Any) -> Any: 

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

155 of/contained the keys of the argument jsonb expression. 

156 

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

158 conformance. 

159 """ 

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

161 

162 def contained_by(self, other: Any) -> Any: 

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

164 keys of the argument jsonb expression. 

165 """ 

166 return self.operate( 

167 CONTAINED_BY, other, result_type=sqltypes.Boolean 

168 ) 

169 

170 def _setup_getitem(self, index: Any) -> Any: 

171 return GETITEM, index, self.type.text_type # type: ignore[attr-defined] # noqa: E501 

172 

173 def defined(self, key: Any) -> Any: 

174 """Boolean expression. Test for presence of a non-NULL value for 

175 the key. Note that the key may be a SQLA expression. 

176 """ 

177 return _HStoreDefinedFunction(self.expr, key) 

178 

179 def delete(self, key: Any) -> Any: 

180 """HStore expression. Returns the contents of this hstore with the 

181 given key deleted. Note that the key may be a SQLA expression. 

182 """ 

183 if isinstance(key, dict): 

184 key = _serialize_hstore(key) 

185 return _HStoreDeleteFunction(self.expr, key) 

186 

187 def slice(self, array: Any) -> Any: 

188 """HStore expression. Returns a subset of an hstore defined by 

189 array of keys. 

190 """ 

191 return _HStoreSliceFunction(self.expr, array) 

192 

193 def keys(self) -> Any: 

194 """Text array expression. Returns array of keys.""" 

195 return _HStoreKeysFunction(self.expr) 

196 

197 def vals(self) -> Any: 

198 """Text array expression. Returns array of values.""" 

199 return _HStoreValsFunction(self.expr) 

200 

201 def array(self) -> Any: 

202 """Text array expression. Returns array of alternating keys and 

203 values. 

204 """ 

205 return _HStoreArrayFunction(self.expr) 

206 

207 def matrix(self) -> Any: 

208 """Text array expression. Returns array of [key, value] pairs.""" 

209 return _HStoreMatrixFunction(self.expr) 

210 

211 comparator_factory = Comparator 

212 

213 def bind_processor(self, dialect: Any) -> Any: 

214 # note that dialect-specific types like that of psycopg and 

215 # psycopg2 will override this method to allow driver-level conversion 

216 # instead, see _PsycopgHStore 

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

218 if isinstance(value, dict): 

219 return _serialize_hstore(value) 

220 else: 

221 return value 

222 

223 return process 

224 

225 def result_processor(self, dialect: Any, coltype: Any) -> Any: 

226 # note that dialect-specific types like that of psycopg and 

227 # psycopg2 will override this method to allow driver-level conversion 

228 # instead, see _PsycopgHStore 

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

230 if value is not None: 

231 return _parse_hstore(value) 

232 else: 

233 return value 

234 

235 return process 

236 

237 

238class hstore(sqlfunc.GenericFunction[_HSTORE_VAL]): 

239 """Construct an hstore value within a SQL expression using the 

240 PostgreSQL ``hstore()`` function. 

241 

242 The :class:`.hstore` function accepts one or two arguments as described 

243 in the PostgreSQL documentation. 

244 

245 E.g.:: 

246 

247 from sqlalchemy.dialects.postgresql import array, hstore 

248 

249 select(hstore("key1", "value1")) 

250 

251 select( 

252 hstore( 

253 array(["key1", "key2", "key3"]), 

254 array(["value1", "value2", "value3"]), 

255 ) 

256 ) 

257 

258 .. seealso:: 

259 

260 :class:`.HSTORE` - the PostgreSQL ``HSTORE`` datatype. 

261 

262 """ 

263 

264 type = HSTORE() 

265 name = "hstore" 

266 inherit_cache = True 

267 

268 

269class _HStoreDefinedFunction(sqlfunc.GenericFunction[bool]): 

270 type = sqltypes.Boolean() 

271 name = "defined" 

272 inherit_cache = True 

273 

274 

275class _HStoreDeleteFunction(sqlfunc.GenericFunction[_HSTORE_VAL]): 

276 type = HSTORE() 

277 name = "delete" 

278 inherit_cache = True 

279 

280 

281class _HStoreSliceFunction(sqlfunc.GenericFunction[_HSTORE_VAL]): 

282 type = HSTORE() 

283 name = "slice" 

284 inherit_cache = True 

285 

286 

287class _HStoreKeysFunction(sqlfunc.GenericFunction[Any]): 

288 type = ARRAY(sqltypes.Text) 

289 name = "akeys" 

290 inherit_cache = True 

291 

292 

293class _HStoreValsFunction(sqlfunc.GenericFunction[Any]): 

294 type = ARRAY(sqltypes.Text) 

295 name = "avals" 

296 inherit_cache = True 

297 

298 

299class _HStoreArrayFunction(sqlfunc.GenericFunction[Any]): 

300 type = ARRAY(sqltypes.Text) 

301 name = "hstore_to_array" 

302 inherit_cache = True 

303 

304 

305class _HStoreMatrixFunction(sqlfunc.GenericFunction[Any]): 

306 type = ARRAY(sqltypes.Text) 

307 name = "hstore_to_matrix" 

308 inherit_cache = True 

309 

310 

311# 

312# parsing. note that none of this is used with the psycopg2 backend, 

313# which provides its own native extensions. 

314# 

315 

316# My best guess at the parsing rules of hstore literals, since no formal 

317# grammar is given. This is mostly reverse engineered from PG's input parser 

318# behavior. 

319HSTORE_PAIR_RE = re.compile( 

320 r""" 

321( 

322 "(?P<key> (\\ . | [^"\\])* )" # Quoted key 

323) 

324[ ]* => [ ]* # Pair operator, optional adjoining whitespace 

325( 

326 (?P<value_null> NULL ) # NULL value 

327 | "(?P<value> (\\ . | [^"\\])* )" # Quoted value 

328) 

329""", 

330 re.VERBOSE, 

331) 

332 

333HSTORE_DELIMITER_RE = re.compile( 

334 r""" 

335[ ]* , [ ]* 

336""", 

337 re.VERBOSE, 

338) 

339 

340 

341def _parse_error(hstore_str: str, pos: int) -> str: 

342 """format an unmarshalling error.""" 

343 

344 ctx = 20 

345 hslen = len(hstore_str) 

346 

347 parsed_tail = hstore_str[max(pos - ctx - 1, 0) : min(pos, hslen)] 

348 residual = hstore_str[min(pos, hslen) : min(pos + ctx + 1, hslen)] 

349 

350 if len(parsed_tail) > ctx: 

351 parsed_tail = "[...]" + parsed_tail[1:] 

352 if len(residual) > ctx: 

353 residual = residual[:-1] + "[...]" 

354 

355 return "After %r, could not parse residual at position %d: %r" % ( 

356 parsed_tail, 

357 pos, 

358 residual, 

359 ) 

360 

361 

362def _parse_hstore(hstore_str: str) -> _HSTORE_VAL: 

363 """Parse an hstore from its literal string representation. 

364 

365 Attempts to approximate PG's hstore input parsing rules as closely as 

366 possible. Although currently this is not strictly necessary, since the 

367 current implementation of hstore's output syntax is stricter than what it 

368 accepts as input, the documentation makes no guarantees that will always 

369 be the case. 

370 

371 

372 

373 """ 

374 result: _HSTORE_VAL = {} 

375 pos = 0 

376 pair_match = HSTORE_PAIR_RE.match(hstore_str) 

377 

378 while pair_match is not None: 

379 key = pair_match.group("key").replace(r"\"", '"').replace("\\\\", "\\") 

380 if pair_match.group("value_null"): 

381 value = None 

382 else: 

383 value = ( 

384 pair_match.group("value") 

385 .replace(r"\"", '"') 

386 .replace("\\\\", "\\") 

387 ) 

388 result[key] = value 

389 

390 pos += pair_match.end() 

391 

392 delim_match = HSTORE_DELIMITER_RE.match(hstore_str[pos:]) 

393 if delim_match is not None: 

394 pos += delim_match.end() 

395 

396 pair_match = HSTORE_PAIR_RE.match(hstore_str[pos:]) 

397 

398 if pos != len(hstore_str): 

399 raise ValueError(_parse_error(hstore_str, pos)) 

400 

401 return result 

402 

403 

404def _serialize_hstore(val: _HSTORE_VAL) -> str: 

405 """Serialize a dictionary into an hstore literal. Keys and values must 

406 both be strings (except None for values). 

407 

408 """ 

409 

410 def esc(s: Optional[str], position: str) -> str: 

411 if position == "value" and s is None: 

412 return "NULL" 

413 elif isinstance(s, str): 

414 return '"%s"' % s.replace("\\", "\\\\").replace('"', r"\"") 

415 else: 

416 raise ValueError( 

417 "%r in %s position is not a string." % (s, position) 

418 ) 

419 

420 return ", ".join( 

421 "%s=>%s" % (esc(k, "key"), esc(v, "value")) for k, v in val.items() 

422 )