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

1# Copyright (C) 2005-2023 the SQLAlchemy authors and contributors 

2# <see AUTHORS file> 

3# 

4# This module is part of SQLAlchemy and is released under 

5# the MIT License: https://www.opensource.org/licenses/mit-license.php 

6 

7from ... import util 

8from ...sql import coercions 

9from ...sql import roles 

10from ...sql.base import _exclusive_against 

11from ...sql.base import _generative 

12from ...sql.base import ColumnCollection 

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

14from ...sql.elements import ClauseElement 

15from ...sql.expression import alias 

16from ...util.langhelpers import public_factory 

17 

18 

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

20 

21 

22class Insert(StandardInsert): 

23 """SQLite-specific implementation of INSERT. 

24 

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

26 

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

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

29 

30 .. versionadded:: 1.4 

31 

32 .. seealso:: 

33 

34 :ref:`sqlite_on_conflict_insert` 

35 

36 """ 

37 

38 stringify_dialect = "sqlite" 

39 inherit_cache = False 

40 

41 @util.memoized_property 

42 def excluded(self): 

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

44 

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

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

47 all columns in this row to be referenceable. 

48 

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

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

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

52 collection described at :ref:`metadata_tables_and_columns`. 

53 With this collection, ordinary names are accessible like attributes 

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

55 dictionary method names should be accessed using indexed access, 

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

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

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

59 

60 """ 

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

62 

63 _on_conflict_exclusive = _exclusive_against( 

64 "_post_values_clause", 

65 msgs={ 

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

67 "an ON CONFLICT clause established" 

68 }, 

69 ) 

70 

71 @_generative 

72 @_on_conflict_exclusive 

73 def on_conflict_do_update( 

74 self, 

75 index_elements=None, 

76 index_where=None, 

77 set_=None, 

78 where=None, 

79 ): 

80 r""" 

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

82 

83 :param index_elements: 

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

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

86 to infer a target index or unique constraint. 

87 

88 :param index_where: 

89 Additional WHERE criterion that can be used to infer a 

90 conditional target index. 

91 

92 :param set\_: 

93 A dictionary or other mapping object 

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

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

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

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

98 

99 .. versionadded:: 1.4 The 

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

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

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

103 

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

105 Python-specified default UPDATE values or generation functions, 

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

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

108 UPDATE, unless they are manually specified in the 

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

110 

111 :param where: 

112 Optional argument. If present, can be a literal SQL 

113 string or an acceptable expression for a ``WHERE`` clause 

114 that restricts the rows affected by ``DO UPDATE SET``. Rows 

115 not meeting the ``WHERE`` condition will not be updated 

116 (effectively a ``DO NOTHING`` for those rows). 

117 

118 """ 

119 

120 self._post_values_clause = OnConflictDoUpdate( 

121 index_elements, index_where, set_, where 

122 ) 

123 

124 @_generative 

125 @_on_conflict_exclusive 

126 def on_conflict_do_nothing(self, index_elements=None, index_where=None): 

127 """ 

128 Specifies a DO NOTHING action for ON CONFLICT clause. 

129 

130 :param index_elements: 

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

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

133 to infer a target index or unique constraint. 

134 

135 :param index_where: 

136 Additional WHERE criterion that can be used to infer a 

137 conditional target index. 

138 

139 """ 

140 

141 self._post_values_clause = OnConflictDoNothing( 

142 index_elements, index_where 

143 ) 

144 

145 

146insert = public_factory( 

147 Insert, ".dialects.sqlite.insert", ".dialects.sqlite.Insert" 

148) 

149 

150 

151class OnConflictClause(ClauseElement): 

152 stringify_dialect = "sqlite" 

153 

154 def __init__(self, index_elements=None, index_where=None): 

155 

156 if index_elements is not None: 

157 self.constraint_target = None 

158 self.inferred_target_elements = index_elements 

159 self.inferred_target_whereclause = index_where 

160 else: 

161 self.constraint_target = ( 

162 self.inferred_target_elements 

163 ) = self.inferred_target_whereclause = None 

164 

165 

166class OnConflictDoNothing(OnConflictClause): 

167 __visit_name__ = "on_conflict_do_nothing" 

168 

169 

170class OnConflictDoUpdate(OnConflictClause): 

171 __visit_name__ = "on_conflict_do_update" 

172 

173 def __init__( 

174 self, 

175 index_elements=None, 

176 index_where=None, 

177 set_=None, 

178 where=None, 

179 ): 

180 super(OnConflictDoUpdate, self).__init__( 

181 index_elements=index_elements, 

182 index_where=index_where, 

183 ) 

184 

185 if isinstance(set_, dict): 

186 if not set_: 

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

188 elif isinstance(set_, ColumnCollection): 

189 set_ = dict(set_) 

190 else: 

191 raise ValueError( 

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

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

194 "of a Table object" 

195 ) 

196 self.update_values_to_set = [ 

197 (coercions.expect(roles.DMLColumnRole, key), value) 

198 for key, value in set_.items() 

199 ] 

200 self.update_whereclause = where