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

1# dialects/postgresql/ext.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 

8from __future__ import annotations 

9 

10from typing import Any 

11from typing import Iterable 

12from typing import List 

13from typing import Optional 

14from typing import overload 

15from typing import TYPE_CHECKING 

16from typing import TypeVar 

17 

18from . import types 

19from .array import ARRAY 

20from ...sql import coercions 

21from ...sql import elements 

22from ...sql import expression 

23from ...sql import functions 

24from ...sql import roles 

25from ...sql import schema 

26from ...sql.schema import ColumnCollectionConstraint 

27from ...sql.sqltypes import TEXT 

28from ...sql.visitors import InternalTraversal 

29 

30if TYPE_CHECKING: 

31 from ...sql._typing import _ColumnExpressionArgument 

32 from ...sql.elements import ClauseElement 

33 from ...sql.elements import ColumnElement 

34 from ...sql.operators import OperatorType 

35 from ...sql.selectable import FromClause 

36 from ...sql.visitors import _CloneCallableType 

37 from ...sql.visitors import _TraverseInternalsType 

38 

39_T = TypeVar("_T", bound=Any) 

40 

41 

42class aggregate_order_by(expression.ColumnElement[_T]): 

43 """Represent a PostgreSQL aggregate order by expression. 

44 

45 E.g.:: 

46 

47 from sqlalchemy.dialects.postgresql import aggregate_order_by 

48 

49 expr = func.array_agg(aggregate_order_by(table.c.a, table.c.b.desc())) 

50 stmt = select(expr) 

51 

52 would represent the expression: 

53 

54 .. sourcecode:: sql 

55 

56 SELECT array_agg(a ORDER BY b DESC) FROM table; 

57 

58 Similarly:: 

59 

60 expr = func.string_agg( 

61 table.c.a, aggregate_order_by(literal_column("','"), table.c.a) 

62 ) 

63 stmt = select(expr) 

64 

65 Would represent: 

66 

67 .. sourcecode:: sql 

68 

69 SELECT string_agg(a, ',' ORDER BY a) FROM table; 

70 

71 .. versionchanged:: 1.2.13 - the ORDER BY argument may be multiple terms 

72 

73 .. seealso:: 

74 

75 :class:`_functions.array_agg` 

76 

77 """ 

78 

79 __visit_name__ = "aggregate_order_by" 

80 

81 stringify_dialect = "postgresql" 

82 _traverse_internals: _TraverseInternalsType = [ 

83 ("target", InternalTraversal.dp_clauseelement), 

84 ("type", InternalTraversal.dp_type), 

85 ("order_by", InternalTraversal.dp_clauseelement), 

86 ] 

87 

88 @overload 

89 def __init__( 

90 self, 

91 target: ColumnElement[_T], 

92 *order_by: _ColumnExpressionArgument[Any], 

93 ): ... 

94 

95 @overload 

96 def __init__( 

97 self, 

98 target: _ColumnExpressionArgument[_T], 

99 *order_by: _ColumnExpressionArgument[Any], 

100 ): ... 

101 

102 def __init__( 

103 self, 

104 target: _ColumnExpressionArgument[_T], 

105 *order_by: _ColumnExpressionArgument[Any], 

106 ): 

107 self.target: ClauseElement = coercions.expect( 

108 roles.ExpressionElementRole, target 

109 ) 

110 self.type = self.target.type 

111 

112 _lob = len(order_by) 

113 self.order_by: ClauseElement 

114 if _lob == 0: 

115 raise TypeError("at least one ORDER BY element is required") 

116 elif _lob == 1: 

117 self.order_by = coercions.expect( 

118 roles.ExpressionElementRole, order_by[0] 

119 ) 

120 else: 

121 self.order_by = elements.ClauseList( 

122 *order_by, _literal_as_text_role=roles.ExpressionElementRole 

123 ) 

124 

125 def self_group( 

126 self, against: Optional[OperatorType] = None 

127 ) -> ClauseElement: 

128 return self 

129 

130 def get_children(self, **kwargs: Any) -> Iterable[ClauseElement]: 

131 return self.target, self.order_by 

132 

133 def _copy_internals( 

134 self, clone: _CloneCallableType = elements._clone, **kw: Any 

135 ) -> None: 

136 self.target = clone(self.target, **kw) 

137 self.order_by = clone(self.order_by, **kw) 

138 

139 @property 

140 def _from_objects(self) -> List[FromClause]: 

141 return self.target._from_objects + self.order_by._from_objects 

142 

143 

144class ExcludeConstraint(ColumnCollectionConstraint): 

145 """A table-level EXCLUDE constraint. 

146 

147 Defines an EXCLUDE constraint as described in the `PostgreSQL 

148 documentation`__. 

149 

150 __ https://www.postgresql.org/docs/current/static/sql-createtable.html#SQL-CREATETABLE-EXCLUDE 

151 

152 """ # noqa 

153 

154 __visit_name__ = "exclude_constraint" 

155 

156 where = None 

157 inherit_cache = False 

158 

159 create_drop_stringify_dialect = "postgresql" 

160 

161 @elements._document_text_coercion( 

162 "where", 

163 ":class:`.ExcludeConstraint`", 

164 ":paramref:`.ExcludeConstraint.where`", 

165 ) 

166 def __init__(self, *elements, **kw): 

167 r""" 

168 Create an :class:`.ExcludeConstraint` object. 

169 

170 E.g.:: 

171 

172 const = ExcludeConstraint( 

173 (Column("period"), "&&"), 

174 (Column("group"), "="), 

175 where=(Column("group") != "some group"), 

176 ops={"group": "my_operator_class"}, 

177 ) 

178 

179 The constraint is normally embedded into the :class:`_schema.Table` 

180 construct 

181 directly, or added later using :meth:`.append_constraint`:: 

182 

183 some_table = Table( 

184 "some_table", 

185 metadata, 

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

187 Column("period", TSRANGE()), 

188 Column("group", String), 

189 ) 

190 

191 some_table.append_constraint( 

192 ExcludeConstraint( 

193 (some_table.c.period, "&&"), 

194 (some_table.c.group, "="), 

195 where=some_table.c.group != "some group", 

196 name="some_table_excl_const", 

197 ops={"group": "my_operator_class"}, 

198 ) 

199 ) 

200 

201 The exclude constraint defined in this example requires the 

202 ``btree_gist`` extension, that can be created using the 

203 command ``CREATE EXTENSION btree_gist;``. 

204 

205 :param \*elements: 

206 

207 A sequence of two tuples of the form ``(column, operator)`` where 

208 "column" is either a :class:`_schema.Column` object, or a SQL 

209 expression element (e.g. ``func.int8range(table.from, table.to)``) 

210 or the name of a column as string, and "operator" is a string 

211 containing the operator to use (e.g. `"&&"` or `"="`). 

212 

213 In order to specify a column name when a :class:`_schema.Column` 

214 object is not available, while ensuring 

215 that any necessary quoting rules take effect, an ad-hoc 

216 :class:`_schema.Column` or :func:`_expression.column` 

217 object should be used. 

218 The ``column`` may also be a string SQL expression when 

219 passed as :func:`_expression.literal_column` or 

220 :func:`_expression.text` 

221 

222 :param name: 

223 Optional, the in-database name of this constraint. 

224 

225 :param deferrable: 

226 Optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when 

227 issuing DDL for this constraint. 

228 

229 :param initially: 

230 Optional string. If set, emit INITIALLY <value> when issuing DDL 

231 for this constraint. 

232 

233 :param using: 

234 Optional string. If set, emit USING <index_method> when issuing DDL 

235 for this constraint. Defaults to 'gist'. 

236 

237 :param where: 

238 Optional SQL expression construct or literal SQL string. 

239 If set, emit WHERE <predicate> when issuing DDL 

240 for this constraint. 

241 

242 :param ops: 

243 Optional dictionary. Used to define operator classes for the 

244 elements; works the same way as that of the 

245 :ref:`postgresql_ops <postgresql_operator_classes>` 

246 parameter specified to the :class:`_schema.Index` construct. 

247 

248 .. versionadded:: 1.3.21 

249 

250 .. seealso:: 

251 

252 :ref:`postgresql_operator_classes` - general description of how 

253 PostgreSQL operator classes are specified. 

254 

255 """ 

256 columns = [] 

257 render_exprs = [] 

258 self.operators = {} 

259 

260 expressions, operators = zip(*elements) 

261 

262 for (expr, column, strname, add_element), operator in zip( 

263 coercions.expect_col_expression_collection( 

264 roles.DDLConstraintColumnRole, expressions 

265 ), 

266 operators, 

267 ): 

268 if add_element is not None: 

269 columns.append(add_element) 

270 

271 name = column.name if column is not None else strname 

272 

273 if name is not None: 

274 # backwards compat 

275 self.operators[name] = operator 

276 

277 render_exprs.append((expr, name, operator)) 

278 

279 self._render_exprs = render_exprs 

280 

281 ColumnCollectionConstraint.__init__( 

282 self, 

283 *columns, 

284 name=kw.get("name"), 

285 deferrable=kw.get("deferrable"), 

286 initially=kw.get("initially"), 

287 ) 

288 self.using = kw.get("using", "gist") 

289 where = kw.get("where") 

290 if where is not None: 

291 self.where = coercions.expect(roles.StatementOptionRole, where) 

292 

293 self.ops = kw.get("ops", {}) 

294 

295 def _set_parent(self, table, **kw): 

296 super()._set_parent(table) 

297 

298 self._render_exprs = [ 

299 ( 

300 expr if not isinstance(expr, str) else table.c[expr], 

301 name, 

302 operator, 

303 ) 

304 for expr, name, operator in (self._render_exprs) 

305 ] 

306 

307 def _copy(self, target_table=None, **kw): 

308 elements = [ 

309 ( 

310 schema._copy_expression(expr, self.parent, target_table), 

311 operator, 

312 ) 

313 for expr, _, operator in self._render_exprs 

314 ] 

315 c = self.__class__( 

316 *elements, 

317 name=self.name, 

318 deferrable=self.deferrable, 

319 initially=self.initially, 

320 where=self.where, 

321 using=self.using, 

322 ) 

323 c.dispatch._update(self.dispatch) 

324 return c 

325 

326 

327def array_agg(*arg, **kw): 

328 """PostgreSQL-specific form of :class:`_functions.array_agg`, ensures 

329 return type is :class:`_postgresql.ARRAY` and not 

330 the plain :class:`_types.ARRAY`, unless an explicit ``type_`` 

331 is passed. 

332 

333 """ 

334 kw["_default_array_type"] = ARRAY 

335 return functions.func.array_agg(*arg, **kw) 

336 

337 

338class _regconfig_fn(functions.GenericFunction[_T]): 

339 inherit_cache = True 

340 

341 def __init__(self, *args, **kwargs): 

342 args = list(args) 

343 if len(args) > 1: 

344 initial_arg = coercions.expect( 

345 roles.ExpressionElementRole, 

346 args.pop(0), 

347 name=getattr(self, "name", None), 

348 apply_propagate_attrs=self, 

349 type_=types.REGCONFIG, 

350 ) 

351 initial_arg = [initial_arg] 

352 else: 

353 initial_arg = [] 

354 

355 addtl_args = [ 

356 coercions.expect( 

357 roles.ExpressionElementRole, 

358 c, 

359 name=getattr(self, "name", None), 

360 apply_propagate_attrs=self, 

361 ) 

362 for c in args 

363 ] 

364 super().__init__(*(initial_arg + addtl_args), **kwargs) 

365 

366 

367class to_tsvector(_regconfig_fn): 

368 """The PostgreSQL ``to_tsvector`` SQL function. 

369 

370 This function applies automatic casting of the REGCONFIG argument 

371 to use the :class:`_postgresql.REGCONFIG` datatype automatically, 

372 and applies a return type of :class:`_postgresql.TSVECTOR`. 

373 

374 Assuming the PostgreSQL dialect has been imported, either by invoking 

375 ``from sqlalchemy.dialects import postgresql``, or by creating a PostgreSQL 

376 engine using ``create_engine("postgresql...")``, 

377 :class:`_postgresql.to_tsvector` will be used automatically when invoking 

378 ``sqlalchemy.func.to_tsvector()``, ensuring the correct argument and return 

379 type handlers are used at compile and execution time. 

380 

381 .. versionadded:: 2.0.0rc1 

382 

383 """ 

384 

385 inherit_cache = True 

386 type = types.TSVECTOR 

387 

388 

389class to_tsquery(_regconfig_fn): 

390 """The PostgreSQL ``to_tsquery`` SQL function. 

391 

392 This function applies automatic casting of the REGCONFIG argument 

393 to use the :class:`_postgresql.REGCONFIG` datatype automatically, 

394 and applies a return type of :class:`_postgresql.TSQUERY`. 

395 

396 Assuming the PostgreSQL dialect has been imported, either by invoking 

397 ``from sqlalchemy.dialects import postgresql``, or by creating a PostgreSQL 

398 engine using ``create_engine("postgresql...")``, 

399 :class:`_postgresql.to_tsquery` will be used automatically when invoking 

400 ``sqlalchemy.func.to_tsquery()``, ensuring the correct argument and return 

401 type handlers are used at compile and execution time. 

402 

403 .. versionadded:: 2.0.0rc1 

404 

405 """ 

406 

407 inherit_cache = True 

408 type = types.TSQUERY 

409 

410 

411class plainto_tsquery(_regconfig_fn): 

412 """The PostgreSQL ``plainto_tsquery`` SQL function. 

413 

414 This function applies automatic casting of the REGCONFIG argument 

415 to use the :class:`_postgresql.REGCONFIG` datatype automatically, 

416 and applies a return type of :class:`_postgresql.TSQUERY`. 

417 

418 Assuming the PostgreSQL dialect has been imported, either by invoking 

419 ``from sqlalchemy.dialects import postgresql``, or by creating a PostgreSQL 

420 engine using ``create_engine("postgresql...")``, 

421 :class:`_postgresql.plainto_tsquery` will be used automatically when 

422 invoking ``sqlalchemy.func.plainto_tsquery()``, ensuring the correct 

423 argument and return type handlers are used at compile and execution time. 

424 

425 .. versionadded:: 2.0.0rc1 

426 

427 """ 

428 

429 inherit_cache = True 

430 type = types.TSQUERY 

431 

432 

433class phraseto_tsquery(_regconfig_fn): 

434 """The PostgreSQL ``phraseto_tsquery`` SQL function. 

435 

436 This function applies automatic casting of the REGCONFIG argument 

437 to use the :class:`_postgresql.REGCONFIG` datatype automatically, 

438 and applies a return type of :class:`_postgresql.TSQUERY`. 

439 

440 Assuming the PostgreSQL dialect has been imported, either by invoking 

441 ``from sqlalchemy.dialects import postgresql``, or by creating a PostgreSQL 

442 engine using ``create_engine("postgresql...")``, 

443 :class:`_postgresql.phraseto_tsquery` will be used automatically when 

444 invoking ``sqlalchemy.func.phraseto_tsquery()``, ensuring the correct 

445 argument and return type handlers are used at compile and execution time. 

446 

447 .. versionadded:: 2.0.0rc1 

448 

449 """ 

450 

451 inherit_cache = True 

452 type = types.TSQUERY 

453 

454 

455class websearch_to_tsquery(_regconfig_fn): 

456 """The PostgreSQL ``websearch_to_tsquery`` SQL function. 

457 

458 This function applies automatic casting of the REGCONFIG argument 

459 to use the :class:`_postgresql.REGCONFIG` datatype automatically, 

460 and applies a return type of :class:`_postgresql.TSQUERY`. 

461 

462 Assuming the PostgreSQL dialect has been imported, either by invoking 

463 ``from sqlalchemy.dialects import postgresql``, or by creating a PostgreSQL 

464 engine using ``create_engine("postgresql...")``, 

465 :class:`_postgresql.websearch_to_tsquery` will be used automatically when 

466 invoking ``sqlalchemy.func.websearch_to_tsquery()``, ensuring the correct 

467 argument and return type handlers are used at compile and execution time. 

468 

469 .. versionadded:: 2.0.0rc1 

470 

471 """ 

472 

473 inherit_cache = True 

474 type = types.TSQUERY 

475 

476 

477class ts_headline(_regconfig_fn): 

478 """The PostgreSQL ``ts_headline`` SQL function. 

479 

480 This function applies automatic casting of the REGCONFIG argument 

481 to use the :class:`_postgresql.REGCONFIG` datatype automatically, 

482 and applies a return type of :class:`_types.TEXT`. 

483 

484 Assuming the PostgreSQL dialect has been imported, either by invoking 

485 ``from sqlalchemy.dialects import postgresql``, or by creating a PostgreSQL 

486 engine using ``create_engine("postgresql...")``, 

487 :class:`_postgresql.ts_headline` will be used automatically when invoking 

488 ``sqlalchemy.func.ts_headline()``, ensuring the correct argument and return 

489 type handlers are used at compile and execution time. 

490 

491 .. versionadded:: 2.0.0rc1 

492 

493 """ 

494 

495 inherit_cache = True 

496 type = TEXT 

497 

498 def __init__(self, *args, **kwargs): 

499 args = list(args) 

500 

501 # parse types according to 

502 # https://www.postgresql.org/docs/current/textsearch-controls.html#TEXTSEARCH-HEADLINE 

503 if len(args) < 2: 

504 # invalid args; don't do anything 

505 has_regconfig = False 

506 elif ( 

507 isinstance(args[1], elements.ColumnElement) 

508 and args[1].type._type_affinity is types.TSQUERY 

509 ): 

510 # tsquery is second argument, no regconfig argument 

511 has_regconfig = False 

512 else: 

513 has_regconfig = True 

514 

515 if has_regconfig: 

516 initial_arg = coercions.expect( 

517 roles.ExpressionElementRole, 

518 args.pop(0), 

519 apply_propagate_attrs=self, 

520 name=getattr(self, "name", None), 

521 type_=types.REGCONFIG, 

522 ) 

523 initial_arg = [initial_arg] 

524 else: 

525 initial_arg = [] 

526 

527 addtl_args = [ 

528 coercions.expect( 

529 roles.ExpressionElementRole, 

530 c, 

531 name=getattr(self, "name", None), 

532 apply_propagate_attrs=self, 

533 ) 

534 for c in args 

535 ] 

536 super().__init__(*(initial_arg + addtl_args), **kwargs)