Merge pull request #24 from rl-institut/feature/dev-queue

Allow option to send tasks to grid_dev queue

Author: paulapreuss

docker-compose.yml

@@ -42,6 +42,18 @@ services:
       - redis
     networks:
       - sim_network
+  worker-grid-dev:
+    environment:
+      - CELERY_BROKER_URL=redis://redis:6379/0
+      - CELERY_RESULT_BACKEND=redis://redis:6379/0
+      - CELERY_TASK_NAME=grid_dev
+    build:
+      context: .
+      dockerfile: ./task_queue/Dockerfile
+    depends_on:
+      - redis
+    networks:
+      - sim_network
   redis:
     image: redis
     networks:

fastapi_app/webapp.py

@@ -47,10 +47,16 @@
 # Test Driven Development --> https://fastapi.tiangolo.com/tutorial/testing/
 
 def validate_simulation_queue(queue):
-    if queue not in ["grid", "supply"]:
-        raise HTTPException(status_code=400, detail=f'{queue} is not an accepted argument. Simulation options are "grid" or "supply"')
+    if queue not in ["grid", "grid_dev", "supply"]:
+        raise HTTPException(status_code=400, detail=f'{queue} is not an accepted argument. Simulation options are "grid", "grid_dev", or "supply"')
     return
 
+
+def _task_name_for_queue(queue: str) -> str:
+    if queue == "grid_dev":
+        return "grid.run_simulation"
+    return f"{queue}.run_simulation"
+
 @app.get("/")
 def index(request: Request) -> Response:
 
@@ -69,7 +75,7 @@ async def simulate_json_variable(request: Request, queue: str = "supply"):
 
     # send the task to celery
     task = celery_app.send_task(
-        f"{queue}.run_simulation", args=[input_dict], queue=queue, kwargs={}
+        _task_name_for_queue(queue), args=[input_dict], queue=queue, kwargs={}
     )
     queue_answer = await check_task(task.id)
 
@@ -108,7 +114,7 @@ def run_simulation(request: Request, input_json=None, queue="supply") -> Respons
 
     # send the task to celery
     task = celery_app.send_task(
-        f"{queue}.run_simulation", args=[input_dict], queue=queue, kwargs={}
+        _task_name_for_queue(queue), args=[input_dict], queue=queue, kwargs={}
     )
 
     return templates.TemplateResponse(
@@ -150,7 +156,8 @@ async def revoke_task(task_id: str) -> JSONResponse:
 
 @app.get("/schema/{queue}/{variant}")
 def get_schema(queue: str, variant: str):
-    schema_file = f"{queue}_schema"
+    schema_queue = "grid" if queue == "grid_dev" else queue
+    schema_file = f"{schema_queue}_schema"
     schema_path = os.path.join(os.path.dirname(__file__), "static", f"{schema_file}.py")
 
     if not os.path.isfile(schema_path):

fastapi_app/worker.py

@@ -11,5 +11,6 @@
 app = Celery("tasks", broker=CELERY_BROKER_URL, backend=CELERY_RESULT_BACKEND)
 app.conf.task_queues = (
     Queue("grid", routing_key="grid.#"),
+    Queue("grid_dev", routing_key="grid_dev.#"),
     Queue("supply", routing_key="supply.#"),
 )