Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/skeleton/tasks.py: 52%

1import logging 

2import typing as t 

3import logics 

4import time 

5 

6from viur.core import ( 

7 conf, 

8 current, 

9 db, 

10 email, 

11 errors, 

12 tasks, 

13 utils, 

14) 

15from .utils import skeletonByKind, listKnownSkeletons 

16from .relskel import RelSkel 

17 

18from ..bones.raw import RawBone 

19from ..bones.record import RecordBone 

20from ..bones.relational import RelationalBone, RelationalConsistency, RelationalUpdateLevel 

21from ..bones.select import SelectBone 

22from ..bones.string import StringBone 

23 

24 

25@tasks.CallDeferred 

26def update_relations( 

27 key: db.Key, 

28 *, 

29 min_change_time: t.Optional[float] = None, 

30 changed_bones: t.Optional[t.Iterable[str] | str] = (), 

31 cursor: t.Optional[str] = None, 

32 total: int = 0, 

33 **kwargs 

34): 

35 """ 

36 This function updates Entities, which may have a copy of values from another entity which has been recently 

37 edited (updated). In ViUR, relations are implemented by copying the values from the referenced entity into the 

38 entity that's referencing them. This allows ViUR to run queries over properties of referenced entities and 

39 prevents additional db.Get's to these referenced entities if the main entity is read. However, this forces 

40 us to track changes made to entities as we might have to update these mirrored values. This is the deferred 

41 call from meth:`viur.core.skeleton.Skeleton.write()` after an update (edit) on one Entity to do exactly that. 

42 

43 :param key: The database-key of the entity that has been edited 

44 :param min_change_time: The timestamp on which the edit occurred. As we run deferred, and the entity might have 

45 been edited multiple times before we get acutally called, we can ignore entities that have been updated 

46 in the meantime as they're already up-to-date 

47 :param changed_bones: If set, we'll update only entites that have a copy of that bones. Relations mirror only 

48 key and name by default, so we don't have to update these if only another bone has been changed. 

49 :param cursor: The database cursor for the current request as we only process five entities at once and then 

50 defer again. 

51 """ 

52 # TODO: Remove in VIUR4 

53 for _dep, _new in { 

54 "changedBone": "changed_bones", 

55 "minChangeTime": "min_change_time", 

56 "destKey": "key", 

57 }.items(): 

58 if _dep in kwargs: 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true

59 logging.warning(f"{_dep!r} parameter is deprecated, please use {_new!r} instead",) 

60 locals()[_new] = kwargs.pop(_dep) 

61 

62 if min_change_time is None: 62 ↛ 65line 62 didn't jump to line 65 because the condition on line 62 was always true

63 min_change_time = time.time() + 1 

64 

65 changed_bones = utils.ensure_iterable(changed_bones) 

66 

67 if not cursor: 67 ↛ 70line 67 didn't jump to line 70 because the condition on line 67 was always true

68 logging.debug(f"update_relations {key=} {min_change_time=} {changed_bones=}") 

69 

70 if request_data := current.request_data.get(): 70 ↛ 73line 70 didn't jump to line 73 because the condition on line 70 was always true

71 request_data["__update_relations_bones"] = changed_bones 

72 

73 query = db.Query("viur-relations") \ 

74 .filter("dest.__key__ =", key) \ 

75 .filter("viur_delayed_update_tag <", min_change_time) \ 

76 .filter("viur_relational_updateLevel =", RelationalUpdateLevel.Always.value) 

77 

78 if changed_bones: 78 ↛ 81line 78 didn't jump to line 81 because the condition on line 78 was always true

79 query.filter("viur_foreign_keys IN", changed_bones) 

80 

81 query.setCursor(cursor) 

82 

83 # A single entity can reference the same destination more than once (e.g. a list 

84 # bone holding the same reference several times), yielding one relation per 

85 # occurrence. Refreshing that entity once is enough -- doing it per relation only 

86 # stacks transactions on the very same entity and can exceed the request deadline. 

87 seen_src_keys: set[db.Key] = set() 

88 

89 for src_rel in query.run(): 

90 src_key = src_rel["src"].key 

91 if src_key in seen_src_keys: 

92 continue 

93 seen_src_keys.add(src_key) 

94 

95 try: 

96 skel = skeletonByKind(src_rel["viur_src_kind"])() 

97 except AssertionError: 

98 logging.info(f"""Ignoring {src_rel.key!r} which refers to unknown kind {src_rel["viur_src_kind"]!r}""") 

99 continue 

100 

101 try: 

102 skel.patch(lambda skel: skel.refresh(), key=src_key, update_relations=False) 

103 except ValueError: 

104 logging.warning(f"Cannot update stale reference to {src_key!r} referenced by {src_rel.key!r}") 

105 continue 

106 

107 total += 1 

108 

109 if next_cursor := query.getCursor(): 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 update_relations( 

111 key=key, 

112 min_change_time=min_change_time, 

113 changed_bones=changed_bones, 

114 cursor=next_cursor, 

115 total=total 

116 ) 

117 else: 

118 logging.debug(f"update_relations finished with {total=} on {key=} {min_change_time=} {changed_bones=}") 

119 

120 

121class SkelIterTask(tasks.QueryIter): 

122 """ 

123 Iterates the skeletons of a query, and additionally checks a Logics expression. 

124 When the skeleton is valid, it performs the action `data["action"]` on each entry. 

125 """ 

126 

127 @classmethod 

128 def handleEntry(cls, skel, data): 

129 data["total"] += 1 

130 

131 if logics.Logics(data["condition"]).run(skel): 

132 data["count"] += 1 

133 

134 match data["action"]: 

135 case "refresh": 

136 skel.refresh() 

137 if skel["key"]: 

138 skel.write(update_relations=False) 

139 

140 case "delete": 

141 skel.delete() 

142 

143 case other: 

144 assert other == "count" 

145 

146 @classmethod 

147 def handleError(cls, skel, data, exception) -> bool: 

148 logging.exception(exception) 

149 

150 try: 

151 logging.debug(f"{skel=!r}") 

152 except Exception: # noqa 

153 logging.warning("Failed to dump skel") 

154 logging.debug(f"{skel.dbEntity=}") 

155 

156 data["error"] += 1 

157 return True 

158 

159 @classmethod 

160 def handleFinish(cls, total, data): 

161 super().handleFinish(total, data) 

162 

163 if not data["notify"]: 

164 return 

165 

166 txt = ( 

167 f"{conf.instance.project_id}: {data['action']!s} finished for {data['kind']!r}: " 

168 f"{data['count']} of {data['total']}\n" 

169 f"ViUR {data['action']!s}ed {data['count']} skeletons with condition <code>{data['condition']}</code> on a " 

170 f"total of {data['total']} ({data['error']} errored) of kind {data['kind']}.\n" 

171 ) 

172 

173 try: 

174 email.send_email(dests=data["notify"], stringTemplate=txt, skel=None) 

175 except Exception as exc: # noqa; OverQuota, whatever 

176 logging.exception(f'Failed to notify {data["notify"]}') 

177 

178 

179@tasks.CallableTask 

180class SkeletonMaintenanceTask(tasks.CallableTaskBase): 

181 key = "SkeletonMaintenanceTask" 

182 name = "Skeleton Maintenance" 

183 descr = "Perform filtered maintenance operations on skeletons." 

184 

185 def canCall(self): 

186 user = current.user.get() 

187 return user and "root" in user["access"] 

188 

189 class dataSkel(RelSkel): 

190 task = SelectBone( 

191 descr="Task", 

192 required=True, 

193 values={ 

194 "count": "Count", 

195 "refresh": "Refresh (formerly: RebuildSearchIndex)", 

196 "delete": "Delete", 

197 }, 

198 defaultValue="refresh", 

199 ) 

200 

201 kinds = SelectBone( 

202 descr="Kind", 

203 values=listKnownSkeletons, 

204 required=True, 

205 multiple=True, 

206 ) 

207 

208 class FilterRowUsingSkel(RelSkel): 

209 name = StringBone( 

210 required=True, 

211 ) 

212 

213 op = SelectBone( 

214 required=True, 

215 values={ 

216 "$eq": "=", 

217 "$lt": "<", 

218 "$gt": ">", 

219 "$lk": "like", 

220 }, 

221 defaultValue=" ", 

222 ) 

223 

224 value = StringBone( 

225 required=True, 

226 ) 

227 

228 filters = RecordBone( 

229 descr="Filter", 

230 using=FilterRowUsingSkel, 

231 multiple=True, 

232 format="$(name)$(op)=$(value)", 

233 ) 

234 

235 condition = RawBone( 

236 descr="Condition", 

237 type_suffix="code.python", # Logics expression 

238 required=True, 

239 defaultValue="False # fused: by default, doesn't affect anything.\n", 

240 params={ 

241 "tooltip": "Enter a Logics expression here to filter entries by specific skeleton values." 

242 }, 

243 ) 

244 

245 def execute(self, task, kinds, filters, condition): 

246 try: 

247 logics.Logics(condition) 

248 except logics.ParseException as e: 

249 raise errors.BadRequest(f"Error parsing condition {e}") 

250 

251 notify = current.user.get()["name"] 

252 

253 for kind in kinds: 

254 q = skeletonByKind(kind)().all() 

255 

256 for flt in filters: 

257 q.mergeExternalFilter({(flt["name"] + flt["op"]).rstrip("$eq"): flt["value"]}) 

258 

259 params = { 

260 "action": task, 

261 "notify": notify, 

262 "condition": condition, 

263 "kind": kind, 

264 "count": 0, 

265 "total": 0, 

266 "error": 0, 

267 } 

268 

269 SkelIterTask.startIterOnQuery(q, params)