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

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 Dict 

11from typing import List 

12from typing import Optional 

13from typing import Sequence 

14from typing import Union 

15 

16from .._typing import _OnConflictIndexElementsT 

17from .._typing import _OnConflictIndexWhereT 

18from .._typing import _OnConflictSetT 

19from .._typing import _OnConflictWhereT 

20from ... import exc 

21from ... import util 

22from ...sql import coercions 

23from ...sql import roles 

24from ...sql import schema 

25from ...sql._typing import _DMLTableArgument 

26from ...sql.base import ColumnCollection 

27from ...sql.base import ReadOnlyColumnCollection 

28from ...sql.base import SyntaxExtension 

29from ...sql.dml import _DMLColumnElement 

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

31from ...sql.elements import ClauseElement 

32from ...sql.elements import ColumnElement 

33from ...sql.elements import KeyedColumnElement 

34from ...sql.elements import TextClause 

35from ...sql.expression import alias 

36from ...sql.sqltypes import NULLTYPE 

37from ...sql.visitors import InternalTraversal 

38from ...util.typing import Self 

39 

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

41 

42 

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

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

45 construct. 

46 

47 .. container:: inherited_member 

48 

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

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

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

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

53 SQLAlchemy Core. 

54 

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

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

57 :meth:`_sqlite.Insert.on_conflict_do_nothing`. 

58 

59 """ 

60 return Insert(table) 

61 

62 

63class Insert(StandardInsert): 

64 """SQLite-specific implementation of INSERT. 

65 

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

67 

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

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

70 

71 .. versionadded:: 1.4 

72 

73 .. seealso:: 

74 

75 :ref:`sqlite_on_conflict_insert` 

76 

77 """ 

78 

79 stringify_dialect = "sqlite" 

80 inherit_cache = True 

81 

82 @util.memoized_property 

83 def excluded( 

84 self, 

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

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

87 

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

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

90 all columns in this row to be referenceable. 

91 

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

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

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

95 collection described at :ref:`metadata_tables_and_columns`. 

96 With this collection, ordinary names are accessible like attributes 

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

98 dictionary method names should be accessed using indexed access, 

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

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

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

102 

103 """ 

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

105 

106 def on_conflict_do_update( 

107 self, 

108 index_elements: _OnConflictIndexElementsT = None, 

109 index_where: _OnConflictIndexWhereT = None, 

110 set_: _OnConflictSetT = None, 

111 where: _OnConflictWhereT = None, 

112 ) -> Self: 

113 r""" 

114 Specifies a DO UPDATE SET action for ON CONFLICT clause. 

115 

116 This method may be invoked more than once against the same 

117 :class:`_sqlite.Insert` construct, where each ``ON CONFLICT`` clause 

118 renders in the order in which it was established. 

119 

120 .. versionadded:: 2.1 Multiple ``ON CONFLICT`` clauses may be 

121 established on a single :class:`_sqlite.Insert` construct. 

122 

123 .. seealso:: 

124 

125 :ref:`sqlite_on_conflict_multiple` 

126 

127 :param index_elements: 

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

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

130 to infer a target index or unique constraint. 

131 

132 :param index_where: 

133 Additional WHERE criterion that can be used to infer a 

134 conditional target index. 

135 

136 :param set\_: 

137 A dictionary or other mapping object 

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

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

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

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

142 

143 .. versionadded:: 1.4 The 

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

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

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

147 

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

149 Python-specified default UPDATE values or generation functions, 

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

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

152 UPDATE, unless they are manually specified in the 

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

154 

155 :param where: 

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

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

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

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

160 

161 """ 

162 

163 return self.ext( 

164 OnConflictDoUpdate(index_elements, index_where, set_, where) 

165 ) 

166 

167 def on_conflict_do_nothing( 

168 self, 

169 index_elements: _OnConflictIndexElementsT = None, 

170 index_where: _OnConflictIndexWhereT = None, 

171 ) -> Self: 

172 """ 

173 Specifies a DO NOTHING action for ON CONFLICT clause. 

174 

175 This method may be invoked more than once against the same 

176 :class:`_sqlite.Insert` construct, and may be combined with 

177 :meth:`_sqlite.Insert.on_conflict_do_update`, where each 

178 ``ON CONFLICT`` clause renders in the order in which it was 

179 established. As SQLite allows only the last ``ON CONFLICT`` clause 

180 of a statement to omit its conflict target, a call that omits 

181 :paramref:`_sqlite.Insert.on_conflict_do_nothing.index_elements` 

182 must be the last clause established, else 

183 :class:`.InvalidRequestError` is raised. 

184 

185 .. versionadded:: 2.1 Multiple ``ON CONFLICT`` clauses may be 

186 established on a single :class:`_sqlite.Insert` construct. 

187 

188 .. seealso:: 

189 

190 :ref:`sqlite_on_conflict_multiple` 

191 

192 :param index_elements: 

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

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

195 to infer a target index or unique constraint. 

196 

197 :param index_where: 

198 Additional WHERE criterion that can be used to infer a 

199 conditional target index. 

200 

201 """ 

202 

203 return self.ext(OnConflictDoNothing(index_elements, index_where)) 

204 

205 

206class OnConflictClause(SyntaxExtension, ClauseElement): 

207 stringify_dialect = "sqlite" 

208 

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

210 inferred_target_whereclause: Optional[ 

211 Union[ColumnElement[Any], TextClause] 

212 ] 

213 

214 _traverse_internals = [ 

215 ("inferred_target_elements", InternalTraversal.dp_multi_list), 

216 ("inferred_target_whereclause", InternalTraversal.dp_clauseelement), 

217 ] 

218 

219 def __init__( 

220 self, 

221 index_elements: _OnConflictIndexElementsT = None, 

222 index_where: _OnConflictIndexWhereT = None, 

223 ): 

224 if index_elements is not None: 

225 self.inferred_target_elements = [ 

226 coercions.expect(roles.DDLConstraintColumnRole, column) 

227 for column in index_elements 

228 ] 

229 self.inferred_target_whereclause = ( 

230 coercions.expect( 

231 roles.WhereHavingRole, 

232 index_where, 

233 ) 

234 if index_where is not None 

235 else None 

236 ) 

237 else: 

238 self.inferred_target_elements = ( 

239 self.inferred_target_whereclause 

240 ) = None 

241 

242 def _append_to_existing( 

243 self, existing: Sequence[ClauseElement] 

244 ) -> Sequence[ClauseElement]: 

245 if existing: 

246 last = existing[-1] 

247 if ( 

248 isinstance(last, OnConflictClause) 

249 and last.inferred_target_elements is None 

250 ): 

251 raise exc.InvalidRequestError( 

252 "This Insert construct already has an ON CONFLICT " 

253 "clause that omits a conflict target; such a clause " 

254 "must be the last ON CONFLICT clause in the statement" 

255 ) 

256 return [*existing, self] 

257 

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

259 insert_stmt.apply_syntax_extension_point( 

260 self._append_to_existing, "post_values" 

261 ) 

262 

263 

264class OnConflictDoNothing(OnConflictClause): 

265 __visit_name__ = "on_conflict_do_nothing" 

266 

267 inherit_cache = True 

268 

269 

270class OnConflictDoUpdate(OnConflictClause): 

271 __visit_name__ = "on_conflict_do_update" 

272 

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

274 update_whereclause: Optional[ColumnElement[Any]] 

275 

276 _traverse_internals = OnConflictClause._traverse_internals + [ 

277 ("update_values_to_set", InternalTraversal.dp_dml_values), 

278 ("update_whereclause", InternalTraversal.dp_clauseelement), 

279 ] 

280 

281 def __init__( 

282 self, 

283 index_elements: _OnConflictIndexElementsT = None, 

284 index_where: _OnConflictIndexWhereT = None, 

285 set_: _OnConflictSetT = None, 

286 where: _OnConflictWhereT = None, 

287 ): 

288 super().__init__( 

289 index_elements=index_elements, 

290 index_where=index_where, 

291 ) 

292 

293 if isinstance(set_, dict): 

294 if not set_: 

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

296 elif isinstance(set_, ColumnCollection): 

297 set_ = dict(set_) 

298 else: 

299 raise ValueError( 

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

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

302 "of a Table object" 

303 ) 

304 self.update_values_to_set = { 

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

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

307 ) 

308 for k, v in set_.items() 

309 } 

310 self.update_whereclause = ( 

311 coercions.expect(roles.WhereHavingRole, where) 

312 if where is not None 

313 else None 

314 )