Create and configure a Celery app, ensuring that the broker is available.
(broker_url, result_backend, max_retries=5, wait_seconds=5)
| 33 | |
| 34 | |
| 35 | def create_celery_app(broker_url, result_backend, max_retries=5, wait_seconds=5): |
| 36 | """ |
| 37 | Create and configure a Celery app, ensuring that the broker is available. |
| 38 | """ |
| 39 | # Wait for the RabbitMQ broker to be ready |
| 40 | for _ in range(max_retries): |
| 41 | try: |
| 42 | # Try to establish a connection to the broker |
| 43 | celery_temp = Celery(broker=broker_url) |
| 44 | celery_temp.connection().ensure_connection(max_retries=1) |
| 45 | print("Successfully connected to the broker!") |
| 46 | break |
| 47 | except OperationalError: |
| 48 | print(f"Broker connection failed. Retrying in {wait_seconds} seconds...") |
| 49 | time.sleep(wait_seconds) |
| 50 | else: |
| 51 | raise Exception("Failed to connect to the broker after several attempts.") |
| 52 | |
| 53 | # Create the Celery app |
| 54 | celery_app = Celery("worker", broker=broker_url, result_backend=result_backend) |
| 55 | celery_app.conf.task_routes = {"celery_worker.test_celery": "celery"} |
| 56 | celery_app.conf.update(task_track_started=True) |
| 57 | |
| 58 | return celery_app |
| 59 | |
| 60 | # Define your broker and result backend URLs |
| 61 | broker_url = AppConfig.CELERY_BROKER_URL |