codekingpro/portable-devtools
114k
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 List
11from typing import Optional
12from typing import Tuple
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 _generative
28from ...sql.base import ColumnCollection
29from ...sql.base import ReadOnlyColumnCollection
30from ...sql.dml import Insert as StandardInsert
31from ...sql.elements import ClauseElement
32from ...sql.elements import ColumnElement
33from ...sql.elements import KeyedColumnElement
34from ...sql.elements import TextClause
35from ...sql.expression import alias
36from ...util.typing import Self
37
38
39__all__ = ("Insert", "insert")
40
41
42def insert(table: _DMLTableArgument) -> Insert:
43 """Construct a PostgreSQL-specific variant :class:`_postgresql.Insert`
44 construct.
45
46 .. container:: inherited_member
47
48 The :func:`sqlalchemy.dialects.postgresql.insert` function creates
49 a :class:`sqlalchemy.dialects.postgresql.Insert`. This class is based
50 on the dialect-agnostic :class:`_sql.Insert` construct which may
51 be constructed using the :func:`_sql.insert` function in
52 SQLAlchemy Core.
53
54 The :class:`_postgresql.Insert` construct includes additional methods
55 :meth:`_postgresql.Insert.on_conflict_do_update`,
56 :meth:`_postgresql.Insert.on_conflict_do_nothing`.
57
58 """
59 return Insert(table)
60
61
62class Insert(StandardInsert):
63 """PostgreSQL-specific implementation of INSERT.
64
65 Adds methods for PG-specific syntaxes such as ON CONFLICT.
66
67 The :class:`_postgresql.Insert` object is created using the
68 :func:`sqlalchemy.dialects.postgresql.insert` function.
69
70 """
71
72 stringify_dialect = "postgresql"
73 inherit_cache = False
74
75 @util.memoized_property
76 def excluded(
77 self,
78 ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]:
79 """Provide the ``excluded`` namespace for an ON CONFLICT statement
80
81 PG's ON CONFLICT clause allows reference to the row that would
82 be inserted, known as ``excluded``. This attribute provides
83 all columns in this row to be referenceable.
84
85 .. tip:: The :attr:`_postgresql.Insert.excluded` attribute is an
86 instance of :class:`_expression.ColumnCollection`, which provides
87 an interface the same as that of the :attr:`_schema.Table.c`
88 collection described at :ref:`metadata_tables_and_columns`.
89 With this collection, ordinary names are accessible like attributes
90 (e.g. ``stmt.excluded.some_column``), but special names and
91 dictionary method names should be accessed using indexed access,
92 such as ``stmt.excluded["column name"]`` or
93 ``stmt.excluded["values"]``. See the docstring for
94 :class:`_expression.ColumnCollection` for further examples.
95
96 .. seealso::
97
98 :ref:`postgresql_insert_on_conflict` - example of how
99 to use :attr:`_expression.Insert.excluded`
100
101 """
102 return alias(self.table, name="excluded").columns
103
104 _on_conflict_exclusive = _exclusive_against(
105 "_post_values_clause",
106 msgs={
107 "_post_values_clause": "This Insert construct already has "
108 "an ON CONFLICT clause established"
109 },
110 )
111
112 @_generative
113 @_on_conflict_exclusive
114 def on_conflict_do_update(
115 self,
116 constraint: _OnConflictConstraintT = None,
117 index_elements: _OnConflictIndexElementsT = None,
118 index_where: _OnConflictIndexWhereT = None,
119 set_: _OnConflictSetT = None,
120 where: _OnConflictWhereT = None,
121 ) -> Self:
122 r"""
123 Specifies a DO UPDATE SET action for ON CONFLICT clause.
124
125 Either the ``constraint`` or ``index_elements`` argument is
126 required, but only one of these can be specified.
127
128 :param constraint:
129 The name of a unique or exclusion constraint on the table,
130 or the constraint object itself if it has a .name attribute.
131
132 :param index_elements:
133 A sequence consisting of string column names, :class:`_schema.Column`
134 objects, or other column expression objects that will be used
135 to infer a target index.
136
137 :param index_where:
138 Additional WHERE criterion that can be used to infer a
139 conditional target index.
140
141 :param set\_:
142 A dictionary or other mapping object
143 where the keys are either names of columns in the target table,
144 or :class:`_schema.Column` objects or other ORM-mapped columns
145 matching that of the target table, and expressions or literals
146 as values, specifying the ``SET`` actions to take.
147
148 .. versionadded:: 1.4 The
149 :paramref:`_postgresql.Insert.on_conflict_do_update.set_`
150 parameter supports :class:`_schema.Column` objects from the target
151 :class:`_schema.Table` as keys.
152
153 .. warning:: This dictionary does **not** take into account
154 Python-specified default UPDATE values or generation functions,
155 e.g. those specified using :paramref:`_schema.Column.onupdate`.
156 These values will not be exercised for an ON CONFLICT style of
157 UPDATE, unless they are manually specified in the
158 :paramref:`.Insert.on_conflict_do_update.set_` dictionary.
159
160 :param where:
161 Optional argument. An expression object representing a ``WHERE``
162 clause that restricts the rows affected by ``DO UPDATE SET``. Rows not
163 meeting the ``WHERE`` condition will not be updated (effectively a
164 ``DO NOTHING`` for those rows).
165
166
167 .. seealso::
168
169 :ref:`postgresql_insert_on_conflict`
170
171 """
172 self._post_values_clause = OnConflictDoUpdate(
173 constraint, index_elements, index_where, set_, where
174 )
175 return self
176
177 @_generative
178 @_on_conflict_exclusive
179 def on_conflict_do_nothing(
180 self,
181 constraint: _OnConflictConstraintT = None,
182 index_elements: _OnConflictIndexElementsT = None,
183 index_where: _OnConflictIndexWhereT = None,
184 ) -> Self:
185 """
186 Specifies a DO NOTHING action for ON CONFLICT clause.
187
188 The ``constraint`` and ``index_elements`` arguments
189 are optional, but only one of these can be specified.
190
191 :param constraint:
192 The name of a unique or exclusion constraint on the table,
193 or the constraint object itself if it has a .name attribute.
194
195 :param index_elements:
196 A sequence consisting of string column names, :class:`_schema.Column`
197 objects, or other column expression objects that will be used
198 to infer a target index.
199
200 :param index_where:
201 Additional WHERE criterion that can be used to infer a
202 conditional target index.
203
204 .. seealso::
205
206 :ref:`postgresql_insert_on_conflict`
207
208 """
209 self._post_values_clause = OnConflictDoNothing(
210 constraint, index_elements, index_where
211 )
212 return self
213
214
215class OnConflictClause(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 def __init__(
225 self,
226 constraint: _OnConflictConstraintT = None,
227 index_elements: _OnConflictIndexElementsT = None,
228 index_where: _OnConflictIndexWhereT = None,
229 ):
230 if constraint is not None:
231 if not isinstance(constraint, str) and isinstance(
232 constraint,
233 (schema.Constraint, ext.ExcludeConstraint),
234 ):
235 constraint = getattr(constraint, "name") or constraint
236
237 if constraint is not None:
238 if index_elements is not None:
239 raise ValueError(
240 "'constraint' and 'index_elements' are mutually exclusive"
241 )
242
243 if isinstance(constraint, str):
244 self.constraint_target = constraint
245 self.inferred_target_elements = None
246 self.inferred_target_whereclause = None
247 elif isinstance(constraint, schema.Index):
248 index_elements = constraint.expressions
249 index_where = constraint.dialect_options["postgresql"].get(
250 "where"
251 )
252 elif isinstance(constraint, ext.ExcludeConstraint):
253 index_elements = constraint.columns
254 index_where = constraint.where
255 else:
256 index_elements = constraint.columns
257 index_where = constraint.dialect_options["postgresql"].get(
258 "where"
259 )
260
261 if index_elements is not None:
262 self.constraint_target = None
263 self.inferred_target_elements = [
264 coercions.expect(roles.DDLConstraintColumnRole, column)
265 for column in index_elements
266 ]
267
268 self.inferred_target_whereclause = (
269 coercions.expect(
270 (
271 roles.StatementOptionRole
272 if isinstance(constraint, ext.ExcludeConstraint)
273 else roles.WhereHavingRole
274 ),
275 index_where,
276 )
277 if index_where is not None
278 else None
279 )
280
281 elif constraint is None:
282 self.constraint_target = self.inferred_target_elements = (
283 self.inferred_target_whereclause
284 ) = None
285
286
287class OnConflictDoNothing(OnConflictClause):
288 __visit_name__ = "on_conflict_do_nothing"
289
290
291class OnConflictDoUpdate(OnConflictClause):
292 __visit_name__ = "on_conflict_do_update"
293
294 update_values_to_set: List[Tuple[Union[schema.Column[Any], str], Any]]
295 update_whereclause: Optional[ColumnElement[Any]]
296
297 def __init__(
298 self,
299 constraint: _OnConflictConstraintT = None,
300 index_elements: _OnConflictIndexElementsT = None,
301 index_where: _OnConflictIndexWhereT = None,
302 set_: _OnConflictSetT = None,
303 where: _OnConflictWhereT = None,
304 ):
305 super().__init__(
306 constraint=constraint,
307 index_elements=index_elements,
308 index_where=index_where,
309 )
310
311 if (
312 self.inferred_target_elements is None
313 and self.constraint_target is None
314 ):
315 raise ValueError(
316 "Either constraint or index_elements, "
317 "but not both, must be specified unless DO NOTHING"
318 )
319
320 if isinstance(set_, dict):
321 if not set_:
322 raise ValueError("set parameter dictionary must not be empty")
323 elif isinstance(set_, ColumnCollection):
324 set_ = dict(set_)
325 else:
326 raise ValueError(
327 "set parameter must be a non-empty dictionary "
328 "or a ColumnCollection such as the `.c.` collection "
329 "of a Table object"
330 )
331 self.update_values_to_set = [
332 (coercions.expect(roles.DMLColumnRole, key), value)
333 for key, value in set_.items()
334 ]
335 self.update_whereclause = (
336 coercions.expect(roles.WhereHavingRole, where)
337 if where is not None
338 else None
339 )
340