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

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# mypy: ignore-errors 

8 

9 

10import re 

11 

12from .array import ARRAY 

13from .operators import CONTAINED_BY 

14from .operators import CONTAINS 

15from .operators import GETITEM 

16from .operators import HAS_ALL 

17from .operators import HAS_ANY 

18from .operators import HAS_KEY 

19from ... import types as sqltypes 

20from ...sql import functions as sqlfunc 

21 

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

23 

24 

25class HSTORE(sqltypes.Indexable, sqltypes.Concatenable, sqltypes.TypeEngine): 

26 """Represent the PostgreSQL HSTORE type. 

27 

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

29 

30 data_table = Table( 

31 "data_table", 

32 metadata, 

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

34 Column("data", HSTORE), 

35 ) 

36 

37 with engine.connect() as conn: 

38 conn.execute( 

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

40 ) 

41 

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

43 

44 * Index operations:: 

45 

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

47 

48 * Containment operations:: 

49 

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

51 

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

53 

54 * Concatenation:: 

55 

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

57 

58 For a full list of special methods see 

59 :class:`.HSTORE.comparator_factory`. 

60 

61 .. container:: topic 

62 

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

64 

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

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

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

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

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

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

71 work:: 

72 

73 from sqlalchemy.ext.mutable import MutableDict 

74 

75 

76 class MyClass(Base): 

77 __tablename__ = "data_table" 

78 

79 id = Column(Integer, primary_key=True) 

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

81 

82 

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

84 

85 # in-place mutation, requires Mutable extension 

86 # in order for the ORM to detect 

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

88 

89 session.commit() 

90 

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

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

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

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

95 

96 .. seealso:: 

97 

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

99 

100 

101 """ # noqa: E501 

102 

103 __visit_name__ = "HSTORE" 

104 hashable = False 

105 text_type = sqltypes.Text() 

106 

107 def __init__(self, text_type=None): 

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

109 

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

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

112 

113 """ 

114 if text_type is not None: 

115 self.text_type = text_type 

116 

117 class Comparator( 

118 sqltypes.Indexable.Comparator, sqltypes.Concatenable.Comparator 

119 ): 

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

121 

122 def has_key(self, other): 

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

124 key may be a SQLA expression. 

125 """ 

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

127 

128 def has_all(self, other): 

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

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

131 

132 def has_any(self, other): 

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

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

135 

136 def contains(self, other, **kwargs): 

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

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

139 

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

141 conformance. 

142 """ 

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

144 

145 def contained_by(self, other): 

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

147 keys of the argument jsonb expression. 

148 """ 

149 return self.operate( 

150 CONTAINED_BY, other, result_type=sqltypes.Boolean 

151 ) 

152 

153 def _setup_getitem(self, index): 

154 return GETITEM, index, self.type.text_type 

155 

156 def defined(self, key): 

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

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

159 """ 

160 return _HStoreDefinedFunction(self.expr, key) 

161 

162 def delete(self, key): 

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

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

165 """ 

166 if isinstance(key, dict): 

167 key = _serialize_hstore(key) 

168 return _HStoreDeleteFunction(self.expr, key) 

169 

170 def slice(self, array): 

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

172 array of keys. 

173 """ 

174 return _HStoreSliceFunction(self.expr, array) 

175 

176 def keys(self): 

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

178 return _HStoreKeysFunction(self.expr) 

179 

180 def vals(self): 

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

182 return _HStoreValsFunction(self.expr) 

183 

184 def array(self): 

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

186 values. 

187 """ 

188 return _HStoreArrayFunction(self.expr) 

189 

190 def matrix(self): 

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

192 return _HStoreMatrixFunction(self.expr) 

193 

194 comparator_factory = Comparator 

195 

196 def bind_processor(self, dialect): 

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

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

199 # instead, see _PsycopgHStore 

200 def process(value): 

201 if isinstance(value, dict): 

202 return _serialize_hstore(value) 

203 else: 

204 return value 

205 

206 return process 

207 

208 def result_processor(self, dialect, coltype): 

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

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

211 # instead, see _PsycopgHStore 

212 def process(value): 

213 if value is not None: 

214 return _parse_hstore(value) 

215 else: 

216 return value 

217 

218 return process 

219 

220 

221class hstore(sqlfunc.GenericFunction): 

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

223 PostgreSQL ``hstore()`` function. 

224 

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

226 in the PostgreSQL documentation. 

227 

228 E.g.:: 

229 

230 from sqlalchemy.dialects.postgresql import array, hstore 

231 

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

233 

234 select( 

235 hstore( 

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

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

238 ) 

239 ) 

240 

241 .. seealso:: 

242 

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

244 

245 """ 

246 

247 type = HSTORE 

248 name = "hstore" 

249 inherit_cache = True 

250 

251 

252class _HStoreDefinedFunction(sqlfunc.GenericFunction): 

253 type = sqltypes.Boolean 

254 name = "defined" 

255 inherit_cache = True 

256 

257 

258class _HStoreDeleteFunction(sqlfunc.GenericFunction): 

259 type = HSTORE 

260 name = "delete" 

261 inherit_cache = True 

262 

263 

264class _HStoreSliceFunction(sqlfunc.GenericFunction): 

265 type = HSTORE 

266 name = "slice" 

267 inherit_cache = True 

268 

269 

270class _HStoreKeysFunction(sqlfunc.GenericFunction): 

271 type = ARRAY(sqltypes.Text) 

272 name = "akeys" 

273 inherit_cache = True 

274 

275 

276class _HStoreValsFunction(sqlfunc.GenericFunction): 

277 type = ARRAY(sqltypes.Text) 

278 name = "avals" 

279 inherit_cache = True 

280 

281 

282class _HStoreArrayFunction(sqlfunc.GenericFunction): 

283 type = ARRAY(sqltypes.Text) 

284 name = "hstore_to_array" 

285 inherit_cache = True 

286 

287 

288class _HStoreMatrixFunction(sqlfunc.GenericFunction): 

289 type = ARRAY(sqltypes.Text) 

290 name = "hstore_to_matrix" 

291 inherit_cache = True 

292 

293 

294# 

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

296# which provides its own native extensions. 

297# 

298 

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

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

301# behavior. 

302HSTORE_PAIR_RE = re.compile( 

303 r""" 

304( 

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

306) 

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

308( 

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

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

311) 

312""", 

313 re.VERBOSE, 

314) 

315 

316HSTORE_DELIMITER_RE = re.compile( 

317 r""" 

318[ ]* , [ ]* 

319""", 

320 re.VERBOSE, 

321) 

322 

323 

324def _parse_error(hstore_str, pos): 

325 """format an unmarshalling error.""" 

326 

327 ctx = 20 

328 hslen = len(hstore_str) 

329 

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

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

332 

333 if len(parsed_tail) > ctx: 

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

335 if len(residual) > ctx: 

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

337 

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

339 parsed_tail, 

340 pos, 

341 residual, 

342 ) 

343 

344 

345def _parse_hstore(hstore_str): 

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

347 

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

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

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

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

352 be the case. 

353 

354 

355 

356 """ 

357 result = {} 

358 pos = 0 

359 pair_match = HSTORE_PAIR_RE.match(hstore_str) 

360 

361 while pair_match is not None: 

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

363 if pair_match.group("value_null"): 

364 value = None 

365 else: 

366 value = ( 

367 pair_match.group("value") 

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

369 .replace("\\\\", "\\") 

370 ) 

371 result[key] = value 

372 

373 pos += pair_match.end() 

374 

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

376 if delim_match is not None: 

377 pos += delim_match.end() 

378 

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

380 

381 if pos != len(hstore_str): 

382 raise ValueError(_parse_error(hstore_str, pos)) 

383 

384 return result 

385 

386 

387def _serialize_hstore(val): 

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

389 both be strings (except None for values). 

390 

391 """ 

392 

393 def esc(s, position): 

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

395 return "NULL" 

396 elif isinstance(s, str): 

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

398 else: 

399 raise ValueError( 

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

401 ) 

402 

403 return ", ".join( 

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

405 )