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

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

7from __future__ import annotations 

8 

9from typing import Any 

10from typing import Dict 

11from typing import List 

12from typing import Optional 

13from typing import Union 

14 

15from . import ext 

16from .._typing import _OnConflictConstraintT 

17from .._typing import _OnConflictIndexElementsT 

18from .._typing import _OnConflictIndexWhereT 

19from .._typing import _OnConflictSetT 

20from .._typing import _OnConflictWhereT 

21from ... import util 

22from ...sql import coercions 

23from ...sql import roles 

24from ...sql import schema 

25from ...sql._typing import _DMLTableArgument 

26from ...sql.base import _exclusive_against 

27from ...sql.base import ColumnCollection 

28from ...sql.base import ReadOnlyColumnCollection 

29from ...sql.base import SyntaxExtension 

30from ...sql.dml import _DMLColumnElement 

31from ...sql.dml import Insert as StandardInsert 

32from ...sql.elements import ClauseElement 

33from ...sql.elements import ColumnElement 

34from ...sql.elements import KeyedColumnElement 

35from ...sql.elements import TextClause 

36from ...sql.expression import alias 

37from ...sql.type_api import NULLTYPE 

38from ...sql.visitors import InternalTraversal 

39from ...util.typing import Self 

40 

41__all__ = ("Insert", "insert") 

42 

43 

44def insert(table: _DMLTableArgument) -> Insert: 

45 """Construct a PostgreSQL-specific variant :class:`_postgresql.Insert` 

46 construct. 

47 

48 .. container:: inherited_member 

49 

50 The :func:`sqlalchemy.dialects.postgresql.insert` function creates 

51 a :class:`sqlalchemy.dialects.postgresql.Insert`. This class is based 

52 on the dialect-agnostic :class:`_sql.Insert` construct which may 

53 be constructed using the :func:`_sql.insert` function in 

54 SQLAlchemy Core. 

55 

56 The :class:`_postgresql.Insert` construct includes additional methods 

57 :meth:`_postgresql.Insert.on_conflict_do_update`, 

58 :meth:`_postgresql.Insert.on_conflict_do_nothing`. 

59 

60 """ 

61 return Insert(table) 

62 

63 

64class Insert(StandardInsert): 

65 """PostgreSQL-specific implementation of INSERT. 

66 

67 Adds methods for PG-specific syntaxes such as ON CONFLICT. 

68 

69 The :class:`_postgresql.Insert` object is created using the 

70 :func:`sqlalchemy.dialects.postgresql.insert` function. 

71 

72 """ 

73 

74 stringify_dialect = "postgresql" 

75 inherit_cache = True 

76 

77 @util.memoized_property 

78 def excluded( 

79 self, 

80 ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]: 

81 """Provide the ``excluded`` namespace for an ON CONFLICT statement 

82 

83 PG's ON CONFLICT clause allows reference to the row that would 

84 be inserted, known as ``excluded``. This attribute provides 

85 all columns in this row to be referenceable. 

86 

87 .. tip:: The :attr:`_postgresql.Insert.excluded` attribute is an 

88 instance of :class:`_expression.ColumnCollection`, which provides 

89 an interface the same as that of the :attr:`_schema.Table.c` 

90 collection described at :ref:`metadata_tables_and_columns`. 

91 With this collection, ordinary names are accessible like attributes 

92 (e.g. ``stmt.excluded.some_column``), but special names and 

93 dictionary method names should be accessed using indexed access, 

94 such as ``stmt.excluded["column name"]`` or 

95 ``stmt.excluded["values"]``. See the docstring for 

96 :class:`_expression.ColumnCollection` for further examples. 

97 

98 .. seealso:: 

99 

100 :ref:`postgresql_insert_on_conflict` - example of how 

101 to use :attr:`_expression.Insert.excluded` 

102 

103 """ 

104 return alias(self.table, name="excluded").columns 

105 

106 _on_conflict_exclusive = _exclusive_against( 

107 "_post_values_clause", 

108 msgs={ 

109 "_post_values_clause": "This Insert construct already has " 

110 "an ON CONFLICT clause established" 

111 }, 

112 ) 

113 

114 @_on_conflict_exclusive 

115 def on_conflict_do_update( 

116 self, 

117 constraint: _OnConflictConstraintT = None, 

118 index_elements: _OnConflictIndexElementsT = None, 

119 index_where: _OnConflictIndexWhereT = None, 

120 set_: _OnConflictSetT = None, 

121 where: _OnConflictWhereT = None, 

122 ) -> Self: 

123 r""" 

124 Specifies a DO UPDATE SET action for ON CONFLICT clause. 

125 

126 Either the ``constraint`` or ``index_elements`` argument is 

127 required, but only one of these can be specified. 

128 

129 :param constraint: 

130 The name of a unique or exclusion constraint on the table, 

131 or the constraint object itself if it has a .name attribute. 

132 

133 :param index_elements: 

134 A sequence consisting of string column names, :class:`_schema.Column` 

135 objects, or other column expression objects that will be used 

136 to infer a target index. 

137 

138 :param index_where: 

139 Additional WHERE criterion that can be used to infer a 

140 conditional target index. 

141 

142 :param set\_: 

143 A dictionary or other mapping object 

144 where the keys are either names of columns in the target table, 

145 or :class:`_schema.Column` objects or other ORM-mapped columns 

146 matching that of the target table, and expressions or literals 

147 as values, specifying the ``SET`` actions to take. 

148 

149 .. versionadded:: 1.4 The 

150 :paramref:`_postgresql.Insert.on_conflict_do_update.set_` 

151 parameter supports :class:`_schema.Column` objects from the target 

152 :class:`_schema.Table` as keys. 

153 

154 .. warning:: This dictionary does **not** take into account 

155 Python-specified default UPDATE values or generation functions, 

156 e.g. those specified using :paramref:`_schema.Column.onupdate`. 

157 These values will not be exercised for an ON CONFLICT style of 

158 UPDATE, unless they are manually specified in the 

159 :paramref:`.Insert.on_conflict_do_update.set_` dictionary. 

160 

161 :param where: 

162 Optional argument. An expression object representing a ``WHERE`` 

163 clause that restricts the rows affected by ``DO UPDATE SET``. Rows not 

164 meeting the ``WHERE`` condition will not be updated (effectively a 

165 ``DO NOTHING`` for those rows). 

166 

167 

168 .. seealso:: 

169 

170 :ref:`postgresql_insert_on_conflict` 

171 

172 """ 

173 return self.ext( 

174 OnConflictDoUpdate( 

175 constraint, index_elements, index_where, set_, where 

176 ) 

177 ) 

178 

179 @_on_conflict_exclusive 

180 def on_conflict_do_nothing( 

181 self, 

182 constraint: _OnConflictConstraintT = None, 

183 index_elements: _OnConflictIndexElementsT = None, 

184 index_where: _OnConflictIndexWhereT = None, 

185 ) -> Self: 

186 """ 

187 Specifies a DO NOTHING action for ON CONFLICT clause. 

188 

189 The ``constraint`` and ``index_elements`` arguments 

190 are optional, but only one of these can be specified. 

191 

192 :param constraint: 

193 The name of a unique or exclusion constraint on the table, 

194 or the constraint object itself if it has a .name attribute. 

195 

196 :param index_elements: 

197 A sequence consisting of string column names, :class:`_schema.Column` 

198 objects, or other column expression objects that will be used 

199 to infer a target index. 

200 

201 :param index_where: 

202 Additional WHERE criterion that can be used to infer a 

203 conditional target index. 

204 

205 .. seealso:: 

206 

207 :ref:`postgresql_insert_on_conflict` 

208 

209 """ 

210 return self.ext( 

211 OnConflictDoNothing(constraint, index_elements, index_where) 

212 ) 

213 

214 

215class OnConflictClause(SyntaxExtension, ClauseElement): 

216 stringify_dialect = "postgresql" 

217 

218 constraint_target: Optional[str] 

219 inferred_target_elements: Optional[List[Union[str, schema.Column[Any]]]] 

220 inferred_target_whereclause: Optional[ 

221 Union[ColumnElement[Any], TextClause] 

222 ] 

223 

224 _traverse_internals = [ 

225 ("constraint_target", InternalTraversal.dp_string), 

226 ("inferred_target_elements", InternalTraversal.dp_multi_list), 

227 ("inferred_target_whereclause", InternalTraversal.dp_clauseelement), 

228 ] 

229 

230 def __init__( 

231 self, 

232 constraint: _OnConflictConstraintT = None, 

233 index_elements: _OnConflictIndexElementsT = None, 

234 index_where: _OnConflictIndexWhereT = None, 

235 ): 

236 if constraint is not None: 

237 if not isinstance(constraint, str) and isinstance( 

238 constraint, 

239 (schema.Constraint, ext.ExcludeConstraint), 

240 ): 

241 constraint = getattr(constraint, "name") or constraint 

242 

243 if constraint is not None: 

244 if index_elements is not None: 

245 raise ValueError( 

246 "'constraint' and 'index_elements' are mutually exclusive" 

247 ) 

248 

249 if isinstance(constraint, str): 

250 self.constraint_target = constraint 

251 self.inferred_target_elements = None 

252 self.inferred_target_whereclause = None 

253 elif isinstance(constraint, schema.Index): 

254 index_elements = constraint.expressions 

255 index_where = constraint.dialect_options["postgresql"].get( 

256 "where" 

257 ) 

258 elif isinstance(constraint, ext.ExcludeConstraint): 

259 index_elements = constraint.columns 

260 index_where = constraint.where 

261 else: 

262 index_elements = constraint.columns 

263 index_where = constraint.dialect_options["postgresql"].get( 

264 "where" 

265 ) 

266 

267 if index_elements is not None: 

268 self.constraint_target = None 

269 self.inferred_target_elements = [ 

270 coercions.expect(roles.DDLConstraintColumnRole, column) 

271 for column in index_elements 

272 ] 

273 

274 self.inferred_target_whereclause = ( 

275 coercions.expect( 

276 ( 

277 roles.StatementOptionRole 

278 if isinstance(constraint, ext.ExcludeConstraint) 

279 else roles.WhereHavingRole 

280 ), 

281 index_where, 

282 ) 

283 if index_where is not None 

284 else None 

285 ) 

286 

287 elif constraint is None: 

288 self.constraint_target = self.inferred_target_elements = ( 

289 self.inferred_target_whereclause 

290 ) = None 

291 

292 def apply_to_insert(self, insert_stmt: StandardInsert) -> None: 

293 insert_stmt.apply_syntax_extension_point( 

294 self.append_replacing_same_type, "post_values" 

295 ) 

296 

297 

298class OnConflictDoNothing(OnConflictClause): 

299 __visit_name__ = "on_conflict_do_nothing" 

300 

301 inherit_cache = True 

302 

303 

304class OnConflictDoUpdate(OnConflictClause): 

305 __visit_name__ = "on_conflict_do_update" 

306 

307 update_values_to_set: Dict[_DMLColumnElement, ColumnElement[Any]] 

308 update_whereclause: Optional[ColumnElement[Any]] 

309 

310 _traverse_internals = OnConflictClause._traverse_internals + [ 

311 ("update_values_to_set", InternalTraversal.dp_dml_values), 

312 ("update_whereclause", InternalTraversal.dp_clauseelement), 

313 ] 

314 

315 def __init__( 

316 self, 

317 constraint: _OnConflictConstraintT = None, 

318 index_elements: _OnConflictIndexElementsT = None, 

319 index_where: _OnConflictIndexWhereT = None, 

320 set_: _OnConflictSetT = None, 

321 where: _OnConflictWhereT = None, 

322 ): 

323 super().__init__( 

324 constraint=constraint, 

325 index_elements=index_elements, 

326 index_where=index_where, 

327 ) 

328 

329 if ( 

330 self.inferred_target_elements is None 

331 and self.constraint_target is None 

332 ): 

333 raise ValueError( 

334 "Either constraint or index_elements, " 

335 "but not both, must be specified unless DO NOTHING" 

336 ) 

337 

338 if isinstance(set_, dict): 

339 if not set_: 

340 raise ValueError("set parameter dictionary must not be empty") 

341 elif isinstance(set_, ColumnCollection): 

342 set_ = dict(set_) 

343 else: 

344 raise ValueError( 

345 "set parameter must be a non-empty dictionary " 

346 "or a ColumnCollection such as the `.c.` collection " 

347 "of a Table object" 

348 ) 

349 

350 self.update_values_to_set = { 

351 coercions.expect(roles.DMLColumnRole, k): coercions.expect( 

352 roles.ExpressionElementRole, v, type_=NULLTYPE, is_crud=True 

353 ) 

354 for k, v in set_.items() 

355 } 

356 self.update_whereclause = ( 

357 coercions.expect(roles.WhereHavingRole, where) 

358 if where is not None 

359 else None 

360 )