diff --git a/airflow/dags/run_queue.py b/airflow/dags/run_queue.py index a8ab096..f78c8e1 100644 --- a/airflow/dags/run_queue.py +++ b/airflow/dags/run_queue.py @@ -1,9 +1,17 @@ from airflow.sdk import dag, task from pendulum import datetime +from celery import Celery +import os + +app = Celery( + 'airflow_client', + broker = os.getenv('CELERY_BROKER_URL'), + backend = os.getenv('CELERY_BACKEND_URL') +) @dag( schedule="@hourly", - start_date=datetime(2026, 2, 2), + start_date=datetime(2026, 2, 3), description="Run Celery queue with RabbitMQ as the broker \ in order to get GitHub data from the GitHub API", tags=["celery_queue"], @@ -13,7 +21,7 @@ def run_queue(): @task def run_the_queue(): - print("hello") + app.send_task("worker.get_github_data", args=[0, 500]) run_the_queue() diff --git a/airflow/docker-compose.override.yml b/airflow/docker-compose.override.yml new file mode 100644 index 0000000..0cf80ef --- /dev/null +++ b/airflow/docker-compose.override.yml @@ -0,0 +1,14 @@ +services: + api-server: + networks: + - etl-shared + scheduler: + networks: + - etl-shared + triggerer: + networks: + - etl-shared + +networks: + etl-shared: + external: true \ No newline at end of file diff --git a/client.py b/client.py index 6b792b6..f9d99ae 100644 --- a/client.py +++ b/client.py @@ -11,7 +11,6 @@ print("Waiting for Celery task to complete") -# test try: print("Getting the result") response = build_repo_chord(total=5000, batch_size=500) diff --git a/docker-compose.yml b/docker-compose.yml index 1b2cf20..d8e5d0f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -18,6 +18,8 @@ services: timeout: 10s retries: 5 start_period: 60s + networks: + - etl-shared redis: image: redis:7-alpine @@ -31,6 +33,8 @@ services: retries: 5 start_period: 30s restart: on-failure + networks: + - etl-shared celery_worker: build: . @@ -56,6 +60,8 @@ services: redis: condition: service_healthy restart: on-failure + networks: + - etl-shared flower: image: mher/flower @@ -69,6 +75,8 @@ services: condition: service_healthy redis: condition: service_healthy + networks: + - etl-shared client: build: . @@ -83,6 +91,8 @@ services: condition: service_healthy celery_worker: condition: service_healthy + networks: + - etl-shared # airflow-init: @@ -130,4 +140,8 @@ services: volumes: rabbitmq_data: - redis_data: \ No newline at end of file + redis_data: + +networks: + etl-shared: + external: true \ No newline at end of file