codekingpro/portable-devtools

1# dialects/mysql/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 Mapping
13from typing import Optional
14from typing import Tuple
15from typing import Union
16
17from ... import exc
18from ... import util
19from ...sql._typing import _DMLTableArgument
20from ...sql.base import _exclusive_against
21from ...sql.base import _generative
22from ...sql.base import ColumnCollection
23from ...sql.base import ReadOnlyColumnCollection
24from ...sql.dml import Insert as StandardInsert
25from ...sql.elements import ClauseElement
26from ...sql.elements import KeyedColumnElement
27from ...sql.expression import alias
28from ...sql.selectable import NamedFromClause
29from ...util.typing import Self
30
31
32__all__ = ("Insert", "insert")
33
34
35def insert(table: _DMLTableArgument) -> Insert:
36    """Construct a MySQL/MariaDB-specific variant :class:`_mysql.Insert`
37    construct.
38
39    .. container:: inherited_member
40
41        The :func:`sqlalchemy.dialects.mysql.insert` function creates
42        a :class:`sqlalchemy.dialects.mysql.Insert`.  This class is based
43        on the dialect-agnostic :class:`_sql.Insert` construct which may
44        be constructed using the :func:`_sql.insert` function in
45        SQLAlchemy Core.
46
47    The :class:`_mysql.Insert` construct includes additional methods
48    :meth:`_mysql.Insert.on_duplicate_key_update`.
49
50    """
51    return Insert(table)
52
53
54class Insert(StandardInsert):
55    """MySQL-specific implementation of INSERT.
56
57    Adds methods for MySQL-specific syntaxes such as ON DUPLICATE KEY UPDATE.
58
59    The :class:`~.mysql.Insert` object is created using the
60    :func:`sqlalchemy.dialects.mysql.insert` function.
61
62    .. versionadded:: 1.2
63
64    """
65
66    stringify_dialect = "mysql"
67    inherit_cache = False
68
69    @property
70    def inserted(
71        self,
72    ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]:
73        """Provide the "inserted" namespace for an ON DUPLICATE KEY UPDATE
74        statement
75
76        MySQL's ON DUPLICATE KEY UPDATE clause allows reference to the row
77        that would be inserted, via a special function called ``VALUES()``.
78        This attribute provides all columns in this row to be referenceable
79        such that they will render within a ``VALUES()`` function inside the
80        ON DUPLICATE KEY UPDATE clause.    The attribute is named ``.inserted``
81        so as not to conflict with the existing
82        :meth:`_expression.Insert.values` method.
83
84        .. tip::  The :attr:`_mysql.Insert.inserted` attribute is an instance
85            of :class:`_expression.ColumnCollection`, which provides an
86            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.inserted.some_column``), but special names and
90            dictionary method names should be accessed using indexed access,
91            such as ``stmt.inserted["column name"]`` or
92            ``stmt.inserted["values"]``.  See the docstring for
93            :class:`_expression.ColumnCollection` for further examples.
94
95        .. seealso::
96
97            :ref:`mysql_insert_on_duplicate_key_update` - example of how
98            to use :attr:`_expression.Insert.inserted`
99
100        """
101        return self.inserted_alias.columns
102
103    @util.memoized_property
104    def inserted_alias(self) -> NamedFromClause:
105        return alias(self.table, name="inserted")
106
107    @_generative
108    @_exclusive_against(
109        "_post_values_clause",
110        msgs={
111            "_post_values_clause": "This Insert construct already "
112            "has an ON DUPLICATE KEY clause present"
113        },
114    )
115    def on_duplicate_key_update(self, *args: _UpdateArg, **kw: Any) -> Self:
116        r"""
117        Specifies the ON DUPLICATE KEY UPDATE clause.
118
119        :param \**kw:  Column keys linked to UPDATE values.  The
120         values may be any SQL expression or supported literal Python
121         values.
122
123        .. warning:: This dictionary does **not** take into account
124           Python-specified default UPDATE values or generation functions,
125           e.g. those specified using :paramref:`_schema.Column.onupdate`.
126           These values will not be exercised for an ON DUPLICATE KEY UPDATE
127           style of UPDATE, unless values are manually specified here.
128
129        :param \*args: As an alternative to passing key/value parameters,
130         a dictionary or list of 2-tuples can be passed as a single positional
131         argument.
132
133         Passing a single dictionary is equivalent to the keyword argument
134         form::
135
136            insert().on_duplicate_key_update({"name": "some name"})
137
138         Passing a list of 2-tuples indicates that the parameter assignments
139         in the UPDATE clause should be ordered as sent, in a manner similar
140         to that described for the :class:`_expression.Update`
141         construct overall
142         in :ref:`tutorial_parameter_ordered_updates`::
143
144            insert().on_duplicate_key_update(
145                [
146                    ("name", "some name"),
147                    ("value", "some value"),
148                ]
149            )
150
151         .. versionchanged:: 1.3 parameters can be specified as a dictionary
152            or list of 2-tuples; the latter form provides for parameter
153            ordering.
154
155
156        .. versionadded:: 1.2
157
158        .. seealso::
159
160            :ref:`mysql_insert_on_duplicate_key_update`
161
162        """
163        if args and kw:
164            raise exc.ArgumentError(
165                "Can't pass kwargs and positional arguments simultaneously"
166            )
167
168        if args:
169            if len(args) > 1:
170                raise exc.ArgumentError(
171                    "Only a single dictionary or list of tuples "
172                    "is accepted positionally."
173                )
174            values = args[0]
175        else:
176            values = kw
177
178        self._post_values_clause = OnDuplicateClause(
179            self.inserted_alias, values
180        )
181        return self
182
183
184class OnDuplicateClause(ClauseElement):
185    __visit_name__ = "on_duplicate_key_update"
186
187    _parameter_ordering: Optional[List[str]] = None
188
189    update: Dict[str, Any]
190    stringify_dialect = "mysql"
191
192    def __init__(
193        self, inserted_alias: NamedFromClause, update: _UpdateArg
194    ) -> None:
195        self.inserted_alias = inserted_alias
196
197        # auto-detect that parameters should be ordered.   This is copied from
198        # Update._proces_colparams(), however we don't look for a special flag
199        # in this case since we are not disambiguating from other use cases as
200        # we are in Update.values().
201        if isinstance(update, list) and (
202            update and isinstance(update[0], tuple)
203        ):
204            self._parameter_ordering = [key for key, value in update]
205            update = dict(update)
206
207        if isinstance(update, dict):
208            if not update:
209                raise ValueError(
210                    "update parameter dictionary must not be empty"
211                )
212        elif isinstance(update, ColumnCollection):
213            update = dict(update)
214        else:
215            raise ValueError(
216                "update parameter must be a non-empty dictionary "
217                "or a ColumnCollection such as the `.c.` collection "
218                "of a Table object"
219            )
220        self.update = update
221
222
223_UpdateArg = Union[
224    Mapping[Any, Any], List[Tuple[str, Any]], ColumnCollection[Any, Any]
225]
226