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

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 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 )