Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/mysql/dml.py: 62%

1# dialects/mysql/dml.py 

2# Copyright (C) 2005-2025 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]