Coverage for app/celery_app.py: 100.00%

1from celery import Celery 

2 

3from .config import settings 

4 

5celery_app = Celery("worker", broker=settings.CELERY_BROKER_URL, include=["app.tasks"]) 

6 

7celery_app.conf.update( 

8 broker_url=settings.CELERY_BROKER_URL, 

9 result_backend=settings.CELERY_RESULT_BACKEND, # Required so task results are retrievable. 

10 task_serializer="json", 

11 accept_content=["json"], 

12 result_serializer="json", 

13 task_track_started=True, 

14 task_acks_late=True, 

15 worker_prefetch_multiplier=1, 

16 worker_send_task_events=True, 

17 task_send_sent_event=True, 

18 result_extended=True, # Store full result metadata (name, args, kwargs). 

19) 

20 

21 

22if __name__ == "__main__": 

23 celery_app.start()