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

1import abc 

2import datetime 

3import functools 

4import logging 

5import os 

6import sys 

7import traceback 

8import typing as t 

9 

10import grpc 

11import requests 

12from google import protobuf 

13from google.cloud import tasks_v2 

14 

15from viur.core import current, db, errors, utils 

16from viur.core.config import conf 

17from viur.core.decorators import exposed, skey 

18from viur.core.module import Module 

19 

20CUSTOM_OBJ = t.TypeVar("CUSTOM_OBJ") # A JSON serializable object 

21 

22 

23class CustomEnvironmentHandler(abc.ABC): 

24 @abc.abstractmethod 

25 def serialize(self) -> CUSTOM_OBJ: 

26 """Serialize custom environment data 

27 

28 This function must not require any parameters and must 

29 return a JSON serializable object with the desired information. 

30 """ 

31 ... 

32 

33 @abc.abstractmethod 

34 def restore(self, obj: CUSTOM_OBJ) -> None: 

35 """Restore custom environment data 

36 

37 This function will receive the object from :meth:`serialize` and should write 

38 the information it contains to the environment of the deferred request. 

39 """ 

40 ... 

41 

42 

43_gaeApp = os.environ.get("GAE_APPLICATION") 

44 

45queueRegion = None 

46if _gaeApp: 46 ↛ 63line 46 didn't jump to line 63 because the condition on line 46 was always true

47 

48 try: 

49 headers = {"Metadata-Flavor": "Google"} 

50 r = requests.get("http://metadata.google.internal/computeMetadata/v1/instance/region", headers=headers) 

51 # r.text should be look like this "projects/(project-number)/region/(region)" 

52 # like so "projects/1234567890/region/europe-west3" 

53 queueRegion = r.text.split("/")[-1] 

54 except Exception as e: # Something went wrong with the Google Metadata Sever we use the old way 

55 logging.warning(f"Can't obtain queueRegion from Google MetaData Server due exception {e=}") 

56 regionPrefix = _gaeApp.split("~")[0] 

57 regionMap = { 

58 "h": "europe-west3", 

59 "e": "europe-west1" 

60 } 

61 queueRegion = regionMap.get(regionPrefix) 

62 

63if not queueRegion and conf.instance.is_dev_server and os.getenv("TASKS_EMULATOR") is None: 63 ↛ 65line 63 didn't jump to line 65 because the condition on line 63 was never true

64 # Probably local development server 

65 logging.warning("Taskqueue disabled, tasks will run inline!") 

66 

67if not conf.instance.is_dev_server or os.getenv("TASKS_EMULATOR") is None: 67 ↛ 70line 67 didn't jump to line 70 because the condition on line 67 was always true

68 taskClient = tasks_v2.CloudTasksClient() 

69else: 

70 taskClient = tasks_v2.CloudTasksClient( 

71 transport=tasks_v2.services.cloud_tasks.transports.CloudTasksGrpcTransport( 

72 channel=grpc.insecure_channel(os.getenv("TASKS_EMULATOR")) 

73 ) 

74 ) 

75 queueRegion = "local" 

76 

77_periodicTasks: dict[str, dict[t.Callable, datetime.timedelta]] = {} 

78_callableTasks = {} 

79_deferred_tasks = {} 

80_startupTasks = [] 

81_appengineServiceIPs = {"10.0.0.1", "0.1.0.1", "0.1.0.2"} 

82 

83 

84class PermanentTaskFailure(Exception): 

85 """Indicates that a task failed, and will never succeed.""" 

86 pass 

87 

88 

89def removePeriodicTask(task: t.Callable) -> None: 

90 """ 

91 Removes a periodic task from the queue. Useful to unqueue an task 

92 that has been inherited from an overridden module. 

93 """ 

94 global _periodicTasks 

95 assert "periodicTaskName" in dir(task), "This is not a periodic task? " 

96 for queueDict in _periodicTasks.values(): 

97 if task in queueDict: 

98 del queueDict[task] 

99 

100 

101class CallableTaskBase: 

102 """ 

103 Base class for user-callable tasks. 

104 Must be subclassed. 

105 """ 

106 key = None # Unique identifier for this task 

107 name = None # Human-Readable name 

108 descr = None # Human-Readable description 

109 kindName = "server-task" 

110 

111 def canCall(self) -> bool: 

112 """ 

113 Checks wherever the current user can execute this task 

114 :returns: bool 

115 """ 

116 return False 

117 

118 def dataSkel(self): 

119 """ 

120 If additional data is needed, return a skeleton-instance here. 

121 These values are then passed to *execute*. 

122 """ 

123 return None 

124 

125 def execute(self): 

126 """ 

127 The actual code that should be run goes here. 

128 """ 

129 raise NotImplementedError() 

130 

131 

132class TaskHandler(Module): 

133 """ 

134 Task Handler. 

135 Handles calling of Tasks (queued and periodic), and performs updatechecks 

136 Do not Modify. Do not Subclass. 

137 """ 

138 adminInfo = None 

139 retryCountWarningThreshold = 25 

140 

141 def findBoundTask(self, task: t.Callable, obj: object, depth: int = 0) -> t.Optional[tuple[t.Callable, object]]: 

142 

143 """ 

144 Tries to locate the instance, this function belongs to. 

145 If it succeeds in finding it, it returns the function and its instance (-> its "self"). 

146 Otherwise, None is returned. 

147 :param task: A callable decorated with @PeriodicTask 

148 :param obj: Object, which will be scanned in the current iteration. 

149 :param depth: Current iteration depth. 

150 """ 

151 if depth > 3 or "periodicTaskName" not in dir(task): # Limit the maximum amount of recursions 

152 return None 

153 for attr in dir(obj): 

154 if attr.startswith("_"): 

155 continue 

156 try: 

157 v = getattr(obj, attr) 

158 except AttributeError: 

159 continue 

160 if callable(v) and "periodicTaskName" in dir(v) and str(v.periodicTaskName) == str(task.periodicTaskName): 

161 return v, obj 

162 if not isinstance(v, str) and not callable(v): 

163 res = self.findBoundTask(task, v, depth + 1) 

164 if res: 

165 return res 

166 return None 

167 

168 @exposed 

169 def queryIter(self, *args, **kwargs): 

170 """ 

171 This processes one chunk of a queryIter (see below). 

172 """ 

173 req = current.request.get().request 

174 self._validate_request() 

175 data = utils.json.loads(req.body) 

176 if data["classID"] not in MetaQueryIter._classCache: 

177 logging.error(f"""Could not continue queryIter - {data["classID"]} not known on this instance""") 

178 MetaQueryIter._classCache[data["classID"]]._qryStep(data) 

179 

180 @exposed 

181 def deferred(self, *args, **kwargs): 

182 """ 

183 This catches one deferred call and routes it to its destination 

184 """ 

185 req = current.request.get().request 

186 self._validate_request() 

187 # Check if the retry count exceeds our warning threshold 

188 retryCount = req.headers.get("X-Appengine-Taskretrycount", None) 

189 if retryCount and int(retryCount) == self.retryCountWarningThreshold: 

190 from viur.core import email 

191 email.send_email_to_admins( 

192 "Deferred task retry counter exceeded warning threshold", 

193 f"""Task {req.headers.get("X-Appengine-Taskname", "")} is retried for the {retryCount}th time.""" 

194 ) 

195 

196 cmd, data = utils.json.loads(req.body) 

197 funcPath, args, kwargs, env = data 

198 if conf.debug.trace: 

199 logging.debug(f"Call task {funcPath} with {cmd=} {args=} {kwargs=} {env=}") 

200 

201 if env: 

202 if "user" in env and env["user"]: 

203 current.session.get()["user"] = env["user"] 

204 # FIXME: We do not have a fully loaded session from the cookie here, 

205 # but only a partial session. 

206 # But we still leave `loaded` on False, which leads to problems. 

207 

208 # Load current user into context variable if user module is there. 

209 if user_mod := getattr(conf.main_app.vi, "user", None): 

210 current.user.set(user_mod.getCurrentUser()) 

211 if "lang" in env and env["lang"]: 

212 current.language.set(env["lang"]) 

213 if "transactionMarker" in env: 

214 marker = db.get(db.Key("viur-transactionmarker", env["transactionMarker"])) 

215 if not marker: 

216 logging.info(f"""Dropping task, transaction {env["transactionMarker"]} did not apply""") 

217 return 

218 else: 

219 logging.info(f"""Executing task, transaction {env["transactionMarker"]} did succeed""") 

220 if "custom" in env and conf.tasks_custom_environment_handler: 

221 # Check if we need to restore additional environmental data 

222 conf.tasks_custom_environment_handler.restore(env["custom"]) 

223 if cmd == "rel": 

224 caller = conf.main_app 

225 pathlist = [x for x in funcPath.split("/") if x] 

226 for currpath in pathlist: 

227 if currpath not in dir(caller): 

228 logging.error(f"Could not resolve {funcPath=} (failed part was {currpath!r})") 

229 return 

230 caller = getattr(caller, currpath) 

231 try: 

232 caller(*args, **kwargs) 

233 except PermanentTaskFailure: 

234 logging.error("PermanentTaskFailure") 

235 except Exception as e: 

236 logging.exception(e) 

237 raise errors.RequestTimeout() # Task-API should retry 

238 elif cmd == "unb": 

239 if funcPath not in _deferred_tasks: 

240 logging.error(f"Missed deferred task {funcPath=} ({args=},{kwargs=})") 

241 # We call the deferred function *directly* (without walking through the mkDeferred lambda), so ensure 

242 # that any hit to another deferred function will defer again 

243 

244 current.request.get().DEFERRED_TASK_CALLED = True 

245 try: 

246 _deferred_tasks[funcPath](*args, **kwargs) 

247 except PermanentTaskFailure: 

248 logging.error("PermanentTaskFailure") 

249 except Exception as e: 

250 logging.exception(e) 

251 raise errors.RequestTimeout() # Task-API should retry 

252 

253 @exposed 

254 def cron(self, cronName="default", *args, **kwargs): 

255 req = current.request.get() 

256 if not conf.instance.is_dev_server: 

257 self._validate_request(require_cron=True, require_taskname=False) 

258 if cronName not in _periodicTasks: 

259 logging.warning(f"Cron request {cronName} doesn't have any tasks") 

260 # We must defer from cron, as tasks will interpret it as a call originating from task-queue - causing deferred 

261 # functions to be called directly, wich causes calls with _countdown etc set to fail. 

262 req.DEFERRED_TASK_CALLED = True 

263 for task, interval in _periodicTasks[cronName].items(): # Call all periodic tasks bound to that queue 

264 periodicTaskName = task.periodicTaskName.lower() 

265 if interval: # Ensure this task doesn't get called to often 

266 lastCall = db.get(db.Key("viur-task-interval", periodicTaskName)) 

267 if lastCall and utils.utcNow() - lastCall["date"] < interval: 

268 logging.debug(f"Task {periodicTaskName!r} has already run recently - skipping.") 

269 continue 

270 res = self.findBoundTask(task, conf.main_app) 

271 try: 

272 if res: # Its bound, call it this way :) 

273 res[0]() 

274 else: 

275 task() # It seems it wasn't bound - call it as a static method 

276 except Exception as e: 

277 logging.error(f"Error calling periodic task {periodicTaskName}") 

278 logging.exception(e) 

279 else: 

280 logging.debug(f"Successfully called task {periodicTaskName}") 

281 if interval: 

282 # Update its last-call timestamp 

283 entry = db.Entity(db.Key("viur-task-interval", periodicTaskName)) 

284 entry["date"] = utils.utcNow() 

285 db.put(entry) 

286 logging.debug("Periodic tasks complete") 

287 

288 def _validate_request( 

289 self, 

290 *, 

291 require_cron: bool = False, 

292 require_taskname: bool = True, 

293 ) -> None: 

294 """ 

295 Validate the header and metadata of a request 

296 

297 If the request is valid, None will be returned. 

298 Otherwise, an exception will be raised. 

299 

300 :param require_taskname: Require "X-AppEngine-TaskName" header 

301 :param require_cron: Require "X-Appengine-Cron" header 

302 """ 

303 req = current.request.get().request 

304 if ( 

305 req.environ.get("HTTP_X_APPENGINE_USER_IP") not in _appengineServiceIPs 

306 and (not conf.instance.is_dev_server or os.getenv("TASKS_EMULATOR") is None) 

307 ): 

308 logging.critical("Detected an attempted XSRF attack. This request did not originate from Task Queue.") 

309 raise errors.Forbidden() 

310 if require_cron and "X-Appengine-Cron" not in req.headers: 

311 logging.critical('Detected an attempted XSRF attack. The header "X-AppEngine-Cron" was not set.') 

312 raise errors.Forbidden() 

313 if require_taskname and "X-AppEngine-TaskName" not in req.headers: 

314 logging.critical('Detected an attempted XSRF attack. The header "X-AppEngine-Taskname" was not set.') 

315 raise errors.Forbidden() 

316 

317 @exposed 

318 def list(self, *args, **kwargs): 

319 """Lists all user-callable tasks which are callable by this user""" 

320 global _callableTasks 

321 

322 from viur.core.skeleton import SkeletonInstance, SkelList, RelSkel 

323 from viur.core.bones import BaseBone, StringBone 

324 

325 class TaskSkel(RelSkel): 

326 key = BaseBone() 

327 name = StringBone() 

328 descr = StringBone() 

329 

330 tasks = SkelList(TaskSkel, *( 

331 SkeletonInstance(TaskSkel, { 

332 "key": task.key, 

333 "name": str(task.name), 

334 "descr": str(task.descr) 

335 }) for task in _callableTasks.values() if task().canCall() 

336 )) 

337 

338 return self.render.list(tasks) 

339 

340 @exposed 

341 @skey(allow_empty=True) 

342 def execute(self, taskID, *, bounce: bool = False, **kwargs): 

343 """Queues a specific task for the next maintenance run""" 

344 global _callableTasks 

345 

346 if taskID in _callableTasks: 

347 task = _callableTasks[taskID]() 

348 else: 

349 return 

350 

351 if not task.canCall(): 

352 raise errors.Unauthorized() 

353 

354 skel = task.dataSkel() 

355 if ( 

356 not kwargs 

357 or not skel.fromClient(kwargs) 

358 or bounce 

359 ): 

360 return self.render.add(skel) 

361 

362 task.execute(**skel.accessedValues) 

363 

364 return self.render.addSuccess(skel) 

365 

366 

367TaskHandler.admin = True 

368TaskHandler.vi = True 

369TaskHandler.html = True 

370 

371 

372# Decorators 

373 

374def retry_n_times(retries: int, email_recipients: None | str | list[str] = None, 

375 tpl: None | str = None) -> t.Callable: 

376 """ 

377 Wrapper for deferred tasks to limit the amount of retries 

378 

379 :param retries: Number of maximum allowed retries 

380 :param email_recipients: Email addresses to which a info should be sent 

381 when the retry limit is reached. 

382 :param tpl: Instead of the standard text, a custom template can be used. 

383 The name of an email template must be specified. 

384 """ 

385 # language=Jinja2 

386 string_template = \ 

387 """Task {{func_name}} failed {{retries}} times 

388 This was the last attempt.<br> 

389 <pre>{{func_module|escape}}.{{func_name|escape}}({{signature|escape}})</pre> 

390 <pre>{{traceback|escape}}</pre>""" 

391 

392 def outer_wrapper(func): 

393 @functools.wraps(func) 

394 def inner_wrapper(*args, **kwargs): 

395 try: 

396 retry_count = int(current.request.get().request.headers.get("X-Appengine-Taskretrycount", -1)) 

397 except AttributeError: 

398 # During warmup current.request is None (at least on local devserver) 

399 retry_count = -1 

400 try: 

401 return func(*args, **kwargs) 

402 except Exception as exc: 

403 logging.exception(f"Task {func.__qualname__} failed: {exc}") 

404 logging.info( 

405 f"This was the {retry_count}. retry." 

406 f"{retries - retry_count} retries remaining. (total = {retries})" 

407 ) 

408 if retry_count < retries: 

409 # Raise the exception to mark this task as failed, so the task queue can retry it. 

410 raise exc 

411 else: 

412 if email_recipients: 

413 args_repr = [repr(arg) for arg in args] 

414 kwargs_repr = [f"{k!s}={v!r}" for k, v in kwargs.items()] 

415 signature = ", ".join(args_repr + kwargs_repr) 

416 try: 

417 from viur.core import email 

418 email.send_email( 

419 dests=email_recipients, 

420 tpl=tpl, 

421 stringTemplate=string_template if tpl is None else string_template, 

422 # The following params provide information for the emails templates 

423 func_name=func.__name__, 

424 func_qualname=func.__qualname__, 

425 func_module=func.__module__, 

426 retries=retries, 

427 args=args, 

428 kwargs=kwargs, 

429 signature=signature, 

430 traceback=traceback.format_exc(), 

431 ) 

432 except Exception: 

433 logging.exception("Failed to send email to %r", email_recipients) 

434 # Mark as permanently failed (could return nothing too) 

435 raise PermanentTaskFailure() 

436 

437 return inner_wrapper 

438 

439 return outer_wrapper 

440 

441 

442def noRetry(f): 

443 """Prevents a deferred Function from being called a second time""" 

444 logging.warning("Use of `@noRetry` is deprecated; Use `@retry_n_times(0)` instead!", stacklevel=2) 

445 return retry_n_times(0)(f) 

446 

447 

448def CallDeferred(func: t.Callable) -> t.Callable: 

449 """ 

450 This is a decorator, which always calls the wrapped method deferred. 

451 

452 The call will be packed and queued into a Cloud Tasks queue. 

453 The Task Queue calls the TaskHandler which executed the wrapped function 

454 with the originally arguments in a different request. 

455 

456 

457 In addition to the arguments for the wrapped methods you can set these: 

458 

459 _queue: Specify the queue in which the task should be pushed. 

460 If no value is given, the queue name set in `conf.tasks_default_queues` 

461 will be used. If the config does not have a value for this task, "default" 

462 is used as the default. The queue must exist (use the queue.yaml). 

463 

464 _countdown: Specify a time in seconds after which the task should be called. 

465 This time is relative to the moment where the wrapped method has been called. 

466 

467 _eta: Instead of a relative _countdown value you can specify a `datetime` 

468 when the task is scheduled to be attempted or retried. 

469 

470 _name: Specify a custom name for the cloud task. Must be unique and can 

471 contain only letters ([A-Za-z]), numbers ([0-9]), hyphens (-), colons (:), or periods (.). 

472 

473 _target_version: Specify a version on which to run this task. 

474 By default, a task will be run on the same version where the wrapped method has been called. 

475 

476 _call_deferred: Calls the @CallDeferred decorated method directly. 

477 This is for example necessary, to call a super method which is decorated with @CallDeferred. 

478 

479 .. code-block:: python 

480 

481 # Example for use of the _call_deferred-parameter 

482 class A(Module): 

483 @CallDeferred 

484 def task(self): 

485 ... 

486 

487 class B(A): 

488 @CallDeferred 

489 def task(self): 

490 super().task(_call_deferred=False) # avoid secondary deferred call 

491 ... 

492 

493 See also: 

494 https://cloud.google.com/python/docs/reference/cloudtasks/latest/google.cloud.tasks_v2.types.Task 

495 https://cloud.google.com/python/docs/reference/cloudtasks/latest/google.cloud.tasks_v2.types.CreateTaskRequest 

496 """ 

497 if "viur_doc_build" in dir(sys): 497 ↛ 498line 497 didn't jump to line 498 because the condition on line 497 was never true

498 return func 

499 

500 __undefinedFlag_ = object() 

501 

502 def make_deferred( 

503 func: t.Callable, 

504 self=__undefinedFlag_, 

505 *args, 

506 _queue: str = None, 

507 _name: str | None = None, 

508 _call_deferred: bool = True, 

509 _target_version: str = conf.instance.app_version, 

510 _eta: datetime.datetime | None = None, 

511 _countdown: int = 0, 

512 **kwargs 

513 ): 

514 if _eta is not None and _countdown: 514 ↛ 515line 514 didn't jump to line 515 because the condition on line 514 was never true

515 raise ValueError("You cannot set the _countdown and _eta argument together!") 

516 if conf.debug.trace: 516 ↛ 517line 516 didn't jump to line 517 because the condition on line 516 was never true

517 logging.debug( 

518 f"make_deferred {func=}, {self=}, {args=}, {kwargs=}, " 

519 f"{_queue=}, {_name=}, {_call_deferred=}, {_target_version=}, {_eta=}, {_countdown=}" 

520 ) 

521 

522 try: 

523 req = current.request.get() 

524 except Exception: # This will fail for warmup requests 

525 req = None 

526 

527 # It's the deferred method which is called from the task queue, this has to be called directly 

528 _call_deferred &= not (req and req.request.headers.get("X-Appengine-Taskretrycount") 

529 and "DEFERRED_TASK_CALLED" not in dir(req)) 

530 

531 if not _call_deferred: 531 ↛ 541line 531 didn't jump to line 541 because the condition on line 531 was always true

532 # An explicit request to run the task right here. Whether a queue can be reached is a 

533 # property of the environment and must not turn this into a deferred call. 

534 if self is __undefinedFlag_: 534 ↛ 535line 534 didn't jump to line 535 because the condition on line 534 was never true

535 return func(*args, **kwargs) 

536 

537 if req is not None: 

538 req.DEFERRED_TASK_CALLED = True 

539 return func(self, *args, **kwargs) 

540 

541 if not queueRegion: 

542 # Run tasks inline 

543 logging.debug(f"{func=} will be executed inline") 

544 

545 @functools.wraps(func) 

546 def task(): 

547 if self is __undefinedFlag_: 

548 return func(*args, **kwargs) 

549 else: 

550 return func(self, *args, **kwargs) 

551 

552 if req: 

553 req.pendingTasks.append(task) # This property only exists on development server! 

554 else: 

555 # Warmup request or something - we have to call it now as we can't defer it :/ 

556 task() 

557 

558 return # Ensure no result gets passed back 

559 

560 else: 

561 try: 

562 if self.__class__.__name__ == "index": 

563 funcPath = func.__name__ 

564 else: 

565 funcPath = f"{self.modulePath}/{func.__name__}" 

566 command = "rel" 

567 except Exception: 

568 funcPath = f"{func.__name__}.{func.__module__}" 

569 if self is not __undefinedFlag_: 

570 args = (self,) + args # Re-append self to args, as this function is (hopefully) unbound 

571 command = "unb" 

572 

573 if _queue is None: 

574 _queue = conf.tasks_default_queues.get( 

575 funcPath, conf.tasks_default_queues.get("__default__", "default") 

576 ) 

577 

578 # Try to preserve the important data from the current environment 

579 try: # We might get called inside a warmup request without session 

580 usr = current.session.get().get("user") 

581 if "password" in usr: 

582 del usr["password"] 

583 except Exception: 

584 usr = None 

585 

586 env = {"user": usr} 

587 

588 try: 

589 env["lang"] = current.language.get() 

590 except AttributeError: # This isn't originating from a normal request 

591 pass 

592 

593 if db.is_in_transaction(): 

594 # We have to ensure transaction guarantees for that task also 

595 env["transactionMarker"] = db.acquire_transaction_success_marker() 

596 # We move that task at least 90 seconds into the future so the transaction has time to settle 

597 _countdown = max(90, _countdown) # Countdown can be set to None 

598 

599 if conf.tasks_custom_environment_handler: 

600 # Check if this project relies on additional environmental variables and serialize them too 

601 env["custom"] = conf.tasks_custom_environment_handler.serialize() 

602 

603 # Create task description 

604 task = tasks_v2.Task( 

605 app_engine_http_request=tasks_v2.AppEngineHttpRequest( 

606 body=utils.json.dumps((command, (funcPath, args, kwargs, env))).encode(), 

607 http_method=tasks_v2.HttpMethod.POST, 

608 relative_uri="/_tasks/deferred", 

609 app_engine_routing=tasks_v2.AppEngineRouting( 

610 version=_target_version, 

611 ), 

612 ), 

613 ) 

614 if _name is not None: 

615 task.name = taskClient.task_path(conf.instance.project_id, queueRegion, _queue, _name) 

616 

617 # Set a schedule time in case eta (absolut) or countdown (relative) was set. 

618 if seconds := _countdown: 

619 _eta = utils.utcNow() + datetime.timedelta(seconds=seconds) 

620 if _eta: 

621 # We must send a Timestamp Protobuf instead of a date-string 

622 timestamp = protobuf.timestamp_pb2.Timestamp() 

623 timestamp.FromDatetime(_eta) 

624 task.schedule_time = timestamp 

625 

626 # Use the client to build and send the task. 

627 parent = taskClient.queue_path(conf.instance.project_id, queueRegion, _queue) 

628 logging.debug(f"{parent=}, {task=}") 

629 taskClient.create_task(tasks_v2.CreateTaskRequest(parent=parent, task=task)) 

630 

631 logging.info(f"Created task {func.__name__}.{func.__module__} with {args=} {kwargs=} {env=}") 

632 

633 global _deferred_tasks 

634 _deferred_tasks[f"{func.__name__}.{func.__module__}"] = func 

635 

636 @functools.wraps(func) 

637 def wrapper(*args, **kwargs): 

638 return make_deferred(func, *args, **kwargs) 

639 

640 return wrapper 

641 

642 

643def callDeferred(func): 

644 """ 

645 Deprecated version of CallDeferred 

646 """ 

647 import logging, warnings 

648 

649 msg = "Use of @callDeferred is deprecated, use @CallDeferred instead." 

650 logging.warning(msg, stacklevel=3) 

651 warnings.warn(msg, stacklevel=3) 

652 

653 return CallDeferred(func) 

654 

655 

656def PeriodicTask(interval: datetime.timedelta | int | float = 0, cronName: str = "default") -> t.Callable: 

657 """ 

658 Decorator to call a function periodically during cron job execution. 

659 

660 Interval defines a lower bound for the call-frequency for the given task, specified as a timedelta. 

661 

662 The true interval of how often cron jobs are being executed is defined in the project's cron.yaml file. 

663 This defaults to 4 hours (see https://github.com/viur-framework/viur-base/blob/main/deploy/cron.yaml). 

664 In case the interval defined here is lower than 4 hours, the task will be fired once every 4 hours anyway. 

665 

666 :param interval: Call at most the given timedelta. 

667 """ 

668 

669 def make_decorator(fn): 

670 nonlocal interval 

671 if fn.__name__.startswith("_"): 671 ↛ 672line 671 didn't jump to line 672 because the condition on line 671 was never true

672 raise RuntimeError("Periodic called methods cannot start with an underscore! " 

673 f"Please rename {fn.__name__!r}") 

674 

675 if cronName not in _periodicTasks: 

676 _periodicTasks[cronName] = {} 

677 

678 if isinstance(interval, (int, float)) and "tasks.periodic.useminutes" in conf.compatibility: 678 ↛ 679line 678 didn't jump to line 679 because the condition on line 678 was never true

679 logging.warning( 

680 f"PeriodicTask assuming {interval=} minutes here. This is changed into seconds in future. " 

681 f"Please use `datetime.timedelta(minutes={interval})` for clarification.", 

682 stacklevel=2, 

683 ) 

684 interval *= 60 

685 

686 _periodicTasks[cronName][fn] = utils.parse.timedelta(interval) 

687 fn.periodicTaskName = f"{fn.__module__}_{fn.__qualname__}".replace(".", "_").lower() 

688 return fn 

689 

690 return make_decorator 

691 

692 

693def CallableTask(fn: t.Callable) -> t.Callable: 

694 """Marks a Class as representing a user-callable Task. 

695 It *should* extend CallableTaskBase and *must* provide 

696 its API 

697 """ 

698 global _callableTasks 

699 _callableTasks[fn.key] = fn 

700 return fn 

701 

702 

703def StartupTask(fn: t.Callable) -> t.Callable: 

704 """ 

705 Functions decorated with this are called shortly at instance startup. 

706 It's *not* guaranteed that they actually run on the instance that just started up! 

707 Wrapped functions must not take any arguments. 

708 """ 

709 global _startupTasks 

710 _startupTasks.append(fn) 

711 return fn 

712 

713 

714@CallDeferred 

715def runStartupTasks(): 

716 """ 

717 Runs all queued startupTasks. 

718 Do not call directly! 

719 """ 

720 global _startupTasks 

721 for st in _startupTasks: 

722 st() 

723 

724 

725class MetaQueryIter(type): 

726 """ 

727 This is the meta class for QueryIters. 

728 Used only to keep track of all subclasses of QueryIter so we can emit the callbacks 

729 on the correct class. 

730 """ 

731 _classCache = {} # Mapping className -> Class 

732 

733 def __init__(cls, name, bases, dct): 

734 MetaQueryIter._classCache[str(cls)] = cls 

735 cls.__classID__ = str(cls) 

736 super(MetaQueryIter, cls).__init__(name, bases, dct) 

737 

738 

739class QueryIter(object, metaclass=MetaQueryIter): 

740 """ 

741 BaseClass to run a database Query and process each entry matched. 

742 This will run each step deferred, so it is possible to process an arbitrary number of entries 

743 without being limited by time or memory. 

744 

745 To use this class create a subclass, override the classmethods handleEntry and handleFinish and then 

746 call startIterOnQuery with an instance of a database Query (and possible some custom data to pass along) 

747 """ 

748 queueName = "default" # Name of the taskqueue we will run on 

749 

750 @classmethod 

751 def startIterOnQuery(cls, query: db.Query, customData: t.Any = None) -> None: 

752 """ 

753 Starts iterating the given query on this class. Will return immediately, the first batch will already 

754 run deferred. 

755 

756 Warning: Any custom data *must* be json-serializable and *must* be passed in customData. You cannot store 

757 any data on this class as each chunk may run on a different instance! 

758 """ 

759 assert not (query._customMultiQueryMerge or query._calculateInternalMultiQueryLimit), \ 

760 "Cannot iter a query with postprocessing" 

761 assert isinstance(query.queries, db.QueryDefinition), "Unsatisfiable query or query with an IN filter" 

762 qryDict = { 

763 "kind": query.kind, 

764 "srcSkel": query.srcSkel.kindName if query.srcSkel is not None else None, 

765 "filters": query.queries.filters, 

766 "orders": [(propName, sortOrder.value) for propName, sortOrder in query.queries.orders], 

767 "startCursor": query.queries.startCursor, 

768 "endCursor": query.queries.endCursor, 

769 "origKind": query.origKind, 

770 "distinct": query.queries.distinct, 

771 "classID": cls.__classID__, 

772 "customData": customData, 

773 "totalCount": 0 

774 } 

775 cls._requeueStep(qryDict) 

776 

777 @classmethod 

778 def _requeueStep(cls, qryDict: dict[str, t.Any]) -> None: 

779 """ 

780 Internal use only. Pushes a new step defined in qryDict to either the taskqueue or append it to 

781 the current request if we are on the local development server. 

782 """ 

783 if not queueRegion: # Run tasks inline - hopefully development server 

784 req = current.request.get() 

785 task = lambda *args, **kwargs: cls._qryStep(qryDict) 

786 if req: 

787 req.pendingTasks.append(task) # < This property will be only exist on development server! 

788 return 

789 taskClient.create_task(tasks_v2.CreateTaskRequest( 

790 parent=taskClient.queue_path(conf.instance.project_id, queueRegion, cls.queueName), 

791 task=tasks_v2.Task( 

792 app_engine_http_request=tasks_v2.AppEngineHttpRequest( 

793 body=utils.json.dumps(qryDict).encode(), 

794 http_method=tasks_v2.HttpMethod.POST, 

795 relative_uri="/_tasks/queryIter", 

796 app_engine_routing=tasks_v2.AppEngineRouting( 

797 version=conf.instance.app_version, 

798 ), 

799 ) 

800 ), 

801 )) 

802 

803 @classmethod 

804 def _qryStep(cls, qryDict: dict[str, t.Any]) -> None: 

805 """ 

806 Internal use only. Processes one block of five entries from the query defined in qryDict and 

807 reschedules the next block. 

808 """ 

809 from viur.core.skeleton import skeletonByKind 

810 

811 qry = db.Query(qryDict["kind"]) 

812 qry.srcSkel = skeletonByKind(qryDict["srcSkel"])() if qryDict["srcSkel"] else None 

813 qry.queries.filters = qryDict["filters"] 

814 qry.queries.orders = [(propName, db.SortOrder(sortOrder)) for propName, sortOrder in qryDict["orders"]] 

815 qry.setCursor(qryDict["startCursor"], qryDict["endCursor"]) 

816 qry.origKind = qryDict["origKind"] 

817 qry.queries.distinct = qryDict["distinct"] 

818