Coverage for baibu/core/tasks.py: 100%

1from celery import shared_task 

2from django.core.cache import cache 

3from django.utils import timezone 

4 

5HEARTBEAT_CACHE_KEY = "core:worker-heartbeat" 

6 

7 

8@shared_task(ignore_result=True) 

9def heartbeat() -> str: 

10 """Record that a worker executed a task. Celery beat schedules this.""" 

11 now = timezone.now().isoformat() 

12 cache.set(HEARTBEAT_CACHE_KEY, now, timeout=None) 

13 return now