codekingpro/portable-devtools
115k
1# dialects/sqlite/dml.py
2# Copyright (C) 2005-2024 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
10
11from .._typing import _OnConflictIndexElementsT
12from .._typing import _OnConflictIndexWhereT
13from .._typing import _OnConflictSetT
14from .._typing import _OnConflictWhereT
15from ... import util
16from ...sql import coercions
17from ...sql import roles
18from ...sql._typing import _DMLTableArgument
19from ...sql.base import _exclusive_against
20from ...sql.base import _generative
21from ...sql.base import ColumnCollection
22from ...sql.base import ReadOnlyColumnCollection
23from ...sql.dml import Insert as StandardInsert
24from ...sql.elements import ClauseElement
25from ...sql.elements import KeyedColumnElement
26from ...sql.expression import alias
27from ...util.typing import Self
28
29__all__ = ("Insert", "insert")
30
31
32def insert(table: _DMLTableArgument) -> Insert:
33 """Construct a sqlite-specific variant :class:`_sqlite.Insert`
34 construct.
35
36 .. container:: inherited_member
37
38 The :func:`sqlalchemy.dialects.sqlite.insert` function creates
39 a :class:`sqlalchemy.dialects.sqlite.Insert`. This class is based
40 on the dialect-agnostic :class:`_sql.Insert` construct which may
41 be constructed using the :func:`_sql.insert` function in
42 SQLAlchemy Core.
43
44 The :class:`_sqlite.Insert` construct includes additional methods
45 :meth:`_sqlite.Insert.on_conflict_do_update`,
46 :meth:`_sqlite.Insert.on_conflict_do_nothing`.
47
48 """
49 return Insert(table)
50
51
52class Insert(StandardInsert):
53 """SQLite-specific implementation of INSERT.
54
55 Adds methods for SQLite-specific syntaxes such as ON CONFLICT.
56
57 The :class:`_sqlite.Insert` object is created using the
58 :func:`sqlalchemy.dialects.sqlite.insert` function.
59
60 .. versionadded:: 1.4
61
62 .. seealso::
63
64 :ref:`sqlite_on_conflict_insert`
65
66 """
67
68 stringify_dialect = "sqlite"
69 inherit_cache = False
70
71 @util.memoized_property
72 def excluded(
73 self,
74 ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]:
75 """Provide the ``excluded`` namespace for an ON CONFLICT statement
76
77 SQLite's ON CONFLICT clause allows reference to the row that would
78 be inserted, known as ``excluded``. This attribute provides
79 all columns in this row to be referenceable.
80
81 .. tip:: The :attr:`_sqlite.Insert.excluded` attribute is an instance
82 of :class:`_expression.ColumnCollection`, which provides an
83 interface the same as that of the :attr:`_schema.Table.c`
84 collection described at :ref:`metadata_tables_and_columns`.
85 With this collection, ordinary names are accessible like attributes
86 (e.g. ``stmt.excluded.some_column``), but special names and
87 dictionary method names should be accessed using indexed access,
88 such as ``stmt.excluded["column name"]`` or
89 ``stmt.excluded["values"]``. See the docstring for
90 :class:`_expression.ColumnCollection` for further examples.
91
92 """
93 return alias(self.table, name="excluded").columns
94
95 _on_conflict_exclusive = _exclusive_against(
96 "_post_values_clause",
97 msgs={
98 "_post_values_clause": "This Insert construct already has "
99 "an ON CONFLICT clause established"
100 },
101 )
102
103 @_generative
104 @_on_conflict_exclusive
105 def on_conflict_do_update(
106 self,
107 index_elements: _OnConflictIndexElementsT = None,
108 index_where: _OnConflictIndexWhereT = None,
109 set_: _OnConflictSetT = None,
110 where: _OnConflictWhereT = None,
111 ) -> Self:
112 r"""
113 Specifies a DO UPDATE SET action for ON CONFLICT clause.
114
115 :param index_elements:
116 A sequence consisting of string column names, :class:`_schema.Column`
117 objects, or other column expression objects that will be used
118 to infer a target index or unique constraint.
119
120 :param index_where:
121 Additional WHERE criterion that can be used to infer a
122 conditional target index.
123
124 :param set\_:
125 A dictionary or other mapping object
126 where the keys are either names of columns in the target table,
127 or :class:`_schema.Column` objects or other ORM-mapped columns
128 matching that of the target table, and expressions or literals
129 as values, specifying the ``SET`` actions to take.
130
131 .. versionadded:: 1.4 The
132 :paramref:`_sqlite.Insert.on_conflict_do_update.set_`
133 parameter supports :class:`_schema.Column` objects from the target
134 :class:`_schema.Table` as keys.
135
136 .. warning:: This dictionary does **not** take into account
137 Python-specified default UPDATE values or generation functions,
138 e.g. those specified using :paramref:`_schema.Column.onupdate`.
139 These values will not be exercised for an ON CONFLICT style of
140 UPDATE, unless they are manually specified in the
141 :paramref:`.Insert.on_conflict_do_update.set_` dictionary.
142
143 :param where:
144 Optional argument. If present, can be a literal SQL
145 string or an acceptable expression for a ``WHERE`` clause
146 that restricts the rows affected by ``DO UPDATE SET``. Rows
147 not meeting the ``WHERE`` condition will not be updated
148 (effectively a ``DO NOTHING`` for those rows).
149
150 """
151
152 self._post_values_clause = OnConflictDoUpdate(
153 index_elements, index_where, set_, where
154 )
155 return self
156
157 @_generative
158 @_on_conflict_exclusive
159 def on_conflict_do_nothing(
160 self,
161 index_elements: _OnConflictIndexElementsT = None,
162 index_where: _OnConflictIndexWhereT = None,
163 ) -> Self:
164 """
165 Specifies a DO NOTHING action for ON CONFLICT clause.
166
167 :param index_elements:
168 A sequence consisting of string column names, :class:`_schema.Column`
169 objects, or other column expression objects that will be used
170 to infer a target index or unique constraint.
171
172 :param index_where:
173 Additional WHERE criterion that can be used to infer a
174 conditional target index.
175
176 """
177
178 self._post_values_clause = OnConflictDoNothing(
179 index_elements, index_where
180 )
181 return self
182
183
184class OnConflictClause(ClauseElement):
185 stringify_dialect = "sqlite"
186
187 constraint_target: None
188 inferred_target_elements: _OnConflictIndexElementsT
189 inferred_target_whereclause: _OnConflictIndexWhereT
190
191 def __init__(
192 self,
193 index_elements: _OnConflictIndexElementsT = None,
194 index_where: _OnConflictIndexWhereT = None,
195 ):
196 if index_elements is not None:
197 self.constraint_target = None
198 self.inferred_target_elements = index_elements
199 self.inferred_target_whereclause = index_where
200 else:
201 self.constraint_target = self.inferred_target_elements = (
202 self.inferred_target_whereclause
203 ) = None
204
205
206class OnConflictDoNothing(OnConflictClause):
207 __visit_name__ = "on_conflict_do_nothing"
208
209
210class OnConflictDoUpdate(OnConflictClause):
211 __visit_name__ = "on_conflict_do_update"
212
213 def __init__(
214 self,
215 index_elements: _OnConflictIndexElementsT = None,
216 index_where: _OnConflictIndexWhereT = None,
217 set_: _OnConflictSetT = None,
218 where: _OnConflictWhereT = None,
219 ):
220 super().__init__(
221 index_elements=index_elements,
222 index_where=index_where,
223 )
224
225 if isinstance(set_, dict):
226 if not set_:
227 raise ValueError("set parameter dictionary must not be empty")
228 elif isinstance(set_, ColumnCollection):
229 set_ = dict(set_)
230 else:
231 raise ValueError(
232 "set parameter must be a non-empty dictionary "
233 "or a ColumnCollection such as the `.c.` collection "
234 "of a Table object"
235 )
236 self.update_values_to_set = [
237 (coercions.expect(roles.DMLColumnRole, key), value)
238 for key, value in set_.items()
239 ]
240 self.update_whereclause = where
241