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

1from celery import shared_task 

2from django.core.exceptions import ObjectDoesNotExist 

3 

4from . import services 

5 

6 

7@shared_task(ignore_result=True) 

8def execute_run(run_id: str) -> str: 

9 run = services.execute(run_id) 

10 return run.status if run else "skipped" 

11 

12 

13@shared_task(ignore_result=True) 

14def tag_conversations() -> int: 

15 from .tagging import tag_due 

16 

17 return tag_due() 

18 

19 

20@shared_task(ignore_result=True) 

21def sweep_runs() -> dict: 

22 return services.sweep_runs() 

23 

24 

25@shared_task(ignore_result=True) 

26def transcribe_audio(clip_id: str) -> str: 

27 try: 

28 clip = services.transcribe(clip_id) 

29 except ObjectDoesNotExist: 

30 # The conversation was deleted while the audio was being transcribed. 

31 return "deleted" 

32 return clip.status if clip else "skipped" 

33 

34 

35@shared_task(ignore_result=True) 

36def sweep_transcriptions() -> dict: 

37 return services.sweep_transcriptions()