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