Coverage for app/venv/lib/python3.14/site-packages/weblate/wladmin/tasks.py: 49%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5from __future__ import annotations 

6 

7from celery.schedules import crontab 

8from django.conf import settings 

9from django.db import transaction 

10from django.utils import timezone 

11 

12from weblate.utils.celery import app 

13from weblate.utils.lock import WeblateLockTimeoutError 

14from weblate.wladmin.models import BackupService, SupportStatus 

15 

16 

17@app.task(trail=False) 

18@transaction.atomic 

19def support_status_update() -> None: 

20 for support in SupportStatus.objects.filter(enabled=True).select_for_update(): 

21 support.refresh() 

22 support.save() 

23 

24 

25@app.task(trail=False) 

26def backup() -> None: 

27 for service in BackupService.objects.filter(enabled=True): 

28 backup_service.delay(service.pk) 

29 

30 

31@app.task(trail=False, autoretry_for=(WeblateLockTimeoutError,)) 

32def backup_service(pk: int) -> None: 

33 try: 

34 service = BackupService.objects.get(pk=pk) 

35 except BackupService.DoesNotExist: 

36 # The service was removed meanwhile 

37 return 

38 service.ensure_init() 

39 service.backup() 

40 service.prune() 

41 today = timezone.now().date() 

42 if today.weekday() == 3: 

43 service.cleanup() 

44 

45 

46@app.on_after_finalize.connect 

47def setup_periodic_tasks(sender, **kwargs) -> None: 

48 # Randomize this per site to avoid all instances hitting server at the same time 

49 minute_to_run = hash(settings.SITE_DOMAIN) % 1440 

50 sender.add_periodic_task( 

51 crontab(hour=minute_to_run // 60, minute=minute_to_run % 60), 

52 support_status_update.s(), 

53 name="support-status-update", 

54 ) 

55 sender.add_periodic_task(crontab(hour=2, minute=0), backup.s(), name="backup")