codekingpro/portable-devtools

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