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

1# dialects/sqlite/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 Optional 

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 ColumnCollection 

26from ...sql.base import ReadOnlyColumnCollection 

27from ...sql.base import SyntaxExtension 

28from ...sql.dml import _DMLColumnElement 

29from ...sql.dml import Insert as StandardInsert 

30from ...sql.elements import ClauseElement 

31from ...sql.elements import ColumnElement 

32from ...sql.elements import KeyedColumnElement 

33from ...sql.elements import TextClause 

34from ...sql.expression import alias 

35from ...sql.sqltypes import NULLTYPE 

36from ...sql.visitors import InternalTraversal 

37from ...util.typing import Self 

38 

39__all__ = ("Insert", "insert") 

40 

41 

42def insert(table: _DMLTableArgument) -> Insert: 

43 """Construct a sqlite-specific variant :class:`_sqlite.Insert` 

44 construct. 

45 

46 .. container:: inherited_member 

47 

48 The :func:`sqlalchemy.dialects.sqlite.insert` function creates 

49 a :class:`sqlalchemy.dialects.sqlite.Insert`. This class is based 

50 on the dialect-agnostic :class:`_sql.Insert` construct which may 

51 be constructed using the :func:`_sql.insert` function in 

52 SQLAlchemy Core. 

53 

54 The :class:`_sqlite.Insert` construct includes additional methods 

55 :meth:`_sqlite.Insert.on_conflict_do_update`, 

56 :meth:`_sqlite.Insert.on_conflict_do_nothing`. 

57 

58 """ 

59 return Insert(table) 

60 

61 

62class Insert(StandardInsert): 

63 """SQLite-specific implementation of INSERT. 

64 

65 Adds methods for SQLite-specific syntaxes such as ON CONFLICT. 

66 

67 The :class:`_sqlite.Insert` object is created using the 

68 :func:`sqlalchemy.dialects.sqlite.insert` function. 

69 

70 .. versionadded:: 1.4 

71 

72 .. seealso:: 

73 

74 :ref:`sqlite_on_conflict_insert` 

75 

76 """ 

77 

78 stringify_dialect = "sqlite" 

79 inherit_cache = True 

80 

81 @util.memoized_property 

82 def excluded( 

83 self, 

84 ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]: 

85 """Provide the ``excluded`` namespace for an ON CONFLICT statement 

86 

87 SQLite's ON CONFLICT clause allows reference to the row that would 

88 be inserted, known as ``excluded``. This attribute provides 

89 all columns in this row to be referenceable. 

90 

91 .. tip:: The :attr:`_sqlite.Insert.excluded` attribute is an instance 

92 of :class:`_expression.ColumnCollection`, which provides an 

93 interface the same as that of the :attr:`_schema.Table.c` 

94 collection described at :ref:`metadata_tables_and_columns`. 

95 With this collection, ordinary names are accessible like attributes 

96 (e.g. ``stmt.excluded.some_column``), but special names and 

97 dictionary method names should be accessed using indexed access, 

98 such as ``stmt.excluded["column name"]`` or 

99 ``stmt.excluded["values"]``. See the docstring for 

100 :class:`_expression.ColumnCollection` for further examples. 

101 

102 """ 

103 return alias(self.table, name="excluded").columns 

104 

105 _on_conflict_exclusive = _exclusive_against( 

106 "_post_values_clause", 

107 msgs={ 

108 "_post_values_clause": "This Insert construct already has " 

109 "an ON CONFLICT clause established" 

110 }, 

111 ) 

112 

113 @_on_conflict_exclusive 

114 def on_conflict_do_update( 

115 self, 

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. 

123 

124 :param index_elements: 

125 A sequence consisting of string column names, :class:`_schema.Column` 

126 objects, or other column expression objects that will be used 

127 to infer a target index or unique constraint. 

128 

129 :param index_where: 

130 Additional WHERE criterion that can be used to infer a 

131 conditional target index. 

132 

133 :param set\_: 

134 A dictionary or other mapping object 

135 where the keys are either names of columns in the target table, 

136 or :class:`_schema.Column` objects or other ORM-mapped columns 

137 matching that of the target table, and expressions or literals 

138 as values, specifying the ``SET`` actions to take. 

139 

140 .. versionadded:: 1.4 The 

141 :paramref:`_sqlite.Insert.on_conflict_do_update.set_` 

142 parameter supports :class:`_schema.Column` objects from the target 

143 :class:`_schema.Table` as keys. 

144 

145 .. warning:: This dictionary does **not** take into account 

146 Python-specified default UPDATE values or generation functions, 

147 e.g. those specified using :paramref:`_schema.Column.onupdate`. 

148 These values will not be exercised for an ON CONFLICT style of 

149 UPDATE, unless they are manually specified in the 

150 :paramref:`.Insert.on_conflict_do_update.set_` dictionary. 

151 

152 :param where: 

153 Optional argument. An expression object representing a ``WHERE`` 

154 clause that restricts the rows affected by ``DO UPDATE SET``. Rows not 

155 meeting the ``WHERE`` condition will not be updated (effectively a 

156 ``DO NOTHING`` for those rows). 

157 

158 """ 

159 

160 return self.ext( 

161 OnConflictDoUpdate(index_elements, index_where, set_, where) 

162 ) 

163 

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 return self.ext(OnConflictDoNothing(index_elements, index_where)) 

185 

186 

187class OnConflictClause(SyntaxExtension, ClauseElement): 

188 stringify_dialect = "sqlite" 

189 

190 inferred_target_elements: Optional[List[Union[str, schema.Column[Any]]]] 

191 inferred_target_whereclause: Optional[ 

192 Union[ColumnElement[Any], TextClause] 

193 ] 

194 

195 _traverse_internals = [ 

196 ("inferred_target_elements", InternalTraversal.dp_multi_list), 

197 ("inferred_target_whereclause", InternalTraversal.dp_clauseelement), 

198 ] 

199 

200 def __init__( 

201 self, 

202 index_elements: _OnConflictIndexElementsT = None, 

203 index_where: _OnConflictIndexWhereT = None, 

204 ): 

205 if index_elements is not None: 

206 self.inferred_target_elements = [ 

207 coercions.expect(roles.DDLConstraintColumnRole, column) 

208 for column in index_elements 

209 ] 

210 self.inferred_target_whereclause = ( 

211 coercions.expect( 

212 roles.WhereHavingRole, 

213 index_where, 

214 ) 

215 if index_where is not None 

216 else None 

217 ) 

218 else: 

219 self.inferred_target_elements = ( 

220 self.inferred_target_whereclause 

221 ) = None 

222 

223 def apply_to_insert(self, insert_stmt: StandardInsert) -> None: 

224 insert_stmt.apply_syntax_extension_point( 

225 self.append_replacing_same_type, "post_values" 

226 ) 

227 

228 

229class OnConflictDoNothing(OnConflictClause): 

230 __visit_name__ = "on_conflict_do_nothing" 

231 

232 inherit_cache = True 

233 

234 

235class OnConflictDoUpdate(OnConflictClause): 

236 __visit_name__ = "on_conflict_do_update" 

237 

238 update_values_to_set: Dict[_DMLColumnElement, ColumnElement[Any]] 

239 update_whereclause: Optional[ColumnElement[Any]] 

240 

241 _traverse_internals = OnConflictClause._traverse_internals + [ 

242 ("update_values_to_set", InternalTraversal.dp_dml_values), 

243 ("update_whereclause", InternalTraversal.dp_clauseelement), 

244 ] 

245 

246 def __init__( 

247 self, 

248 index_elements: _OnConflictIndexElementsT = None, 

249 index_where: _OnConflictIndexWhereT = None, 

250 set_: _OnConflictSetT = None, 

251 where: _OnConflictWhereT = None, 

252 ): 

253 super().__init__( 

254 index_elements=index_elements, 

255 index_where=index_where, 

256 ) 

257 

258 if isinstance(set_, dict): 

259 if not set_: 

260 raise ValueError("set parameter dictionary must not be empty") 

261 elif isinstance(set_, ColumnCollection): 

262 set_ = dict(set_) 

263 else: 

264 raise ValueError( 

265 "set parameter must be a non-empty dictionary " 

266 "or a ColumnCollection such as the `.c.` collection " 

267 "of a Table object" 

268 ) 

269 self.update_values_to_set = { 

270 coercions.expect(roles.DMLColumnRole, k): coercions.expect( 

271 roles.ExpressionElementRole, v, type_=NULLTYPE, is_crud=True 

272 ) 

273 for k, v in set_.items() 

274 } 

275 self.update_whereclause = ( 

276 coercions.expect(roles.WhereHavingRole, where) 

277 if where is not None 

278 else None 

279 )