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-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# 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 

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

24 

25 

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

27 """Represent the PostgreSQL HSTORE type. 

28 

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

30 

31 data_table = Table( 

32 "data_table", 

33 metadata, 

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

35 Column("data", HSTORE), 

36 ) 

37 

38 with engine.connect() as conn: 

39 conn.execute( 

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

41 ) 

42 

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

44 

45 * Index operations:: 

46 

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

48 

49 * Containment operations:: 

50 

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

52 

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

54 

55 * Concatenation:: 

56 

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

58 

59 For a full list of special methods see 

60 :class:`.HSTORE.comparator_factory`. 

61 

62 .. container:: topic 

63 

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

65 

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

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

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

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

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

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

72 work:: 

73 

74 from sqlalchemy.ext.mutable import MutableDict 

75 

76 

77 class MyClass(Base): 

78 __tablename__ = "data_table" 

79 

80 id = Column(Integer, primary_key=True) 

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

82 

83 

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

85 

86 # in-place mutation, requires Mutable extension 

87 # in order for the ORM to detect 

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

89 

90 session.commit() 

91 

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

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

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

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

96 

97 .. seealso:: 

98 

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

100 

101 

102 """ # noqa: E501 

103 

104 __visit_name__ = "HSTORE" 

105 hashable = False 

106 text_type = sqltypes.Text() 

107 

108 def __init__(self, text_type=None): 

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

110 

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

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

113 

114 """ 

115 if text_type is not None: 

116 self.text_type = text_type 

117 

118 class Comparator( 

119 sqltypes.Indexable.Comparator, sqltypes.Concatenable.Comparator 

120 ): 

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

122 

123 def has_key(self, other): 

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

125 key may be a SQLA expression. 

126 """ 

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

128 

129 def has_all(self, other): 

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

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

132 

133 def has_any(self, other): 

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

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

136 

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

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

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

140 

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

142 conformance. 

143 """ 

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

145 

146 def contained_by(self, other): 

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

148 keys of the argument jsonb expression. 

149 """ 

150 return self.operate( 

151 CONTAINED_BY, other, result_type=sqltypes.Boolean 

152 ) 

153 

154 def _setup_getitem(self, index): 

155 return GETITEM, index, self.type.text_type 

156 

157 def defined(self, key): 

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

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

160 """ 

161 return _HStoreDefinedFunction(self.expr, key) 

162 

163 def delete(self, key): 

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

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

166 """ 

167 if isinstance(key, dict): 

168 key = _serialize_hstore(key) 

169 return _HStoreDeleteFunction(self.expr, key) 

170 

171 def slice(self, array): 

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

173 array of keys. 

174 """ 

175 return _HStoreSliceFunction(self.expr, array) 

176 

177 def keys(self): 

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

179 return _HStoreKeysFunction(self.expr) 

180 

181 def vals(self): 

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

183 return _HStoreValsFunction(self.expr) 

184 

185 def array(self): 

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

187 values. 

188 """ 

189 return _HStoreArrayFunction(self.expr) 

190 

191 def matrix(self): 

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

193 return _HStoreMatrixFunction(self.expr) 

194 

195 comparator_factory = Comparator 

196 

197 def bind_processor(self, dialect): 

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

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

200 # instead, see _PsycopgHStore 

201 def process(value): 

202 if isinstance(value, dict): 

203 return _serialize_hstore(value) 

204 else: 

205 return value 

206 

207 return process 

208 

209 def result_processor(self, dialect, coltype): 

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

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

212 # instead, see _PsycopgHStore 

213 def process(value): 

214 if value is not None: 

215 return _parse_hstore(value) 

216 else: 

217 return value 

218 

219 return process 

220 

221 

222class hstore(sqlfunc.GenericFunction): 

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

224 PostgreSQL ``hstore()`` function. 

225 

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

227 in the PostgreSQL documentation. 

228 

229 E.g.:: 

230 

231 from sqlalchemy.dialects.postgresql import array, hstore 

232 

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

234 

235 select( 

236 hstore( 

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

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

239 ) 

240 ) 

241 

242 .. seealso:: 

243 

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

245 

246 """ 

247 

248 type = HSTORE 

249 name = "hstore" 

250 inherit_cache = True 

251 

252 

253class _HStoreDefinedFunction(sqlfunc.GenericFunction): 

254 type = sqltypes.Boolean 

255 name = "defined" 

256 inherit_cache = True 

257 

258 

259class _HStoreDeleteFunction(sqlfunc.GenericFunction): 

260 type = HSTORE 

261 name = "delete" 

262 inherit_cache = True 

263 

264 

265class _HStoreSliceFunction(sqlfunc.GenericFunction): 

266 type = HSTORE 

267 name = "slice" 

268 inherit_cache = True 

269 

270 

271class _HStoreKeysFunction(sqlfunc.GenericFunction): 

272 type = ARRAY(sqltypes.Text) 

273 name = "akeys" 

274 inherit_cache = True 

275 

276 

277class _HStoreValsFunction(sqlfunc.GenericFunction): 

278 type = ARRAY(sqltypes.Text) 

279 name = "avals" 

280 inherit_cache = True 

281 

282 

283class _HStoreArrayFunction(sqlfunc.GenericFunction): 

284 type = ARRAY(sqltypes.Text) 

285 name = "hstore_to_array" 

286 inherit_cache = True 

287 

288 

289class _HStoreMatrixFunction(sqlfunc.GenericFunction): 

290 type = ARRAY(sqltypes.Text) 

291 name = "hstore_to_matrix" 

292 inherit_cache = True 

293 

294 

295# 

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

297# which provides its own native extensions. 

298# 

299 

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

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

302# behavior. 

303HSTORE_PAIR_RE = re.compile( 

304 r""" 

305( 

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

307) 

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

309( 

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

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

312) 

313""", 

314 re.VERBOSE, 

315) 

316 

317HSTORE_DELIMITER_RE = re.compile( 

318 r""" 

319[ ]* , [ ]* 

320""", 

321 re.VERBOSE, 

322) 

323 

324 

325def _parse_error(hstore_str, pos): 

326 """format an unmarshalling error.""" 

327 

328 ctx = 20 

329 hslen = len(hstore_str) 

330 

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

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

333 

334 if len(parsed_tail) > ctx: 

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

336 if len(residual) > ctx: 

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

338 

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

340 parsed_tail, 

341 pos, 

342 residual, 

343 ) 

344 

345 

346def _parse_hstore(hstore_str): 

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

348 

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

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

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

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

353 be the case. 

354 

355 

356 

357 """ 

358 result = {} 

359 pos = 0 

360 pair_match = HSTORE_PAIR_RE.match(hstore_str) 

361 

362 while pair_match is not None: 

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

364 if pair_match.group("value_null"): 

365 value = None 

366 else: 

367 value = ( 

368 pair_match.group("value") 

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

370 .replace("\\\\", "\\") 

371 ) 

372 result[key] = value 

373 

374 pos += pair_match.end() 

375 

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

377 if delim_match is not None: 

378 pos += delim_match.end() 

379 

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

381 

382 if pos != len(hstore_str): 

383 raise ValueError(_parse_error(hstore_str, pos)) 

384 

385 return result 

386 

387 

388def _serialize_hstore(val): 

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

390 both be strings (except None for values). 

391 

392 """ 

393 

394 def esc(s, position): 

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

396 return "NULL" 

397 elif isinstance(s, str): 

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

399 else: 

400 raise ValueError( 

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

402 ) 

403 

404 return ", ".join( 

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

406 )