fix: Fix get command in the result attribute, and add `get_group_ta…

…sks` (#11)

.makim.yaml

@@ -56,18 +56,6 @@ groups:
  run: |
  pytest ${{ args.path }} ${{ args.params }}
 
- example:
- help: Run the example app
- shell: bash
- run: |
- set -ex
- sugar build
- sugar ext restart --options -d
- sleep 5
- cd example/
- celery -A tasks.app worker --loglevel=debug &
- python app.py
-
  setup:
  help: Run the setup for the unit tests
  shell: bash

containers/compose.yaml

@@ -4,13 +4,11 @@ services:
  valkey:
  image: valkey/valkey:7.2.5-alpine
  hostname: valkey
- container_name: valkey
  ports:
  - 6379:6379
 
  # celery:
  # hostname: celery
- # container_name: celery
  # build:
  # context: ..
  # dockerfile: containers/celery/Dockerfile

examples/README.md

@@ -0,0 +1,8 @@
+# Examples of usage of Retsu
+
+This folder contains some examples of usage of Retsu.
+
+## Using redis as a queue between celery tasks
+
+The `redis_queue_between_tasks` folder contains an example about how to create
+extra queues to establish communication across different celery chord tasks.

examples/redis_queue_between_tasks/README.md

@@ -0,0 +1,11 @@
+# How to test
+
+You can run `run.sh` in your terminal, and in the web browser you can try the
+following endpoints:
+
+- http://127.0.0.1:5000/serial/10/20
+- http://127.0.0.1:5000/parallel/10/20
+- http://127.0.0.1:5000/serial/result/[TASK_ID]
+- http://127.0.0.1:5000/parallel/result/[TASK_ID]
+
+Remember to replace `[TASK_ID]` by the desired task id.

example/app.py → examples/redis_queue_between_tasks/app.py

@@ -3,7 +3,8 @@
 import os
 import signal
 
-from typing import Optional
+from time import sleep
+from typing import Any, Optional
 
 from flask import Flask
 from tasks import MyTaskManager
@@ -40,55 +41,84 @@ def api() -> str:
  * parallel
  * status
  * result
- """
+
+ Example of endpoints:
+
+ - http://127.0.0.1:5000/serial/1/2
+ - http://127.0.0.1:5000/parallel/1/2
+ - http://127.0.0.1:5000/serial/result/[TASK_ID]
+ - http://127.0.0.1:5000/serial/status/[TASK_ID]
+ - http://127.0.0.1:5000/parallel/result/[TASK_ID]
+ - http://127.0.0.1:5000/parallel/status/[TASK_ID]
+
+ Remember to replace `[TASK_ID]` by the desired task id.
+ """.replace("\n", "<br/>")
 
  return menu
 
 
 @app.route("/serial/<int:a>/<int:b>")
-def serial(a: int, b: int) -> str:
+def serial(a: int, b: int) -> dict[str, Any]:
  """Define the serial endpoint."""
  task1 = task_manager.get_task("serial")
  key = task1.request(a=a, b=b)
- return f"your task ({key}) is running now, please wait until it is done."
+ return {"message": f"Your task ({key}) is running now"}
 
 
 @app.route("/parallel/<int:a>/<int:b>")
-def parallel(a: int, b: int) -> str:
+def parallel(a: int, b: int) -> dict[str, Any]:
  """Define the parallel endpoint."""
  task2 = task_manager.get_task("parallel")
  key = task2.request(a=a, b=b)
- return f"your task ({key}) is running now, please wait until it is done."
+ return {"message": f"Your task ({key}) is running now"}
 
 
 @app.route("/serial/status/<string:task_id>")
-def serial_status(task_id: str) -> str:
+def serial_status(task_id: str) -> dict[str, Any]:
  """Define serial/status endpoint."""
  task1 = task_manager.get_task("serial")
- _status = task1.status(task_id)
+ _status = task1.result.status(task_id)
  return {"status": _status, "task_id": task_id}
 
 
 @app.route("/parallel/status/<string:task_id>")
-def parallel_status(task_id: str) -> str:
+def parallel_status(task_id: str) -> dict[str, Any]:
  """Define parallel/status endpoint."""
  task2 = task_manager.get_task("parallel")
- _status = task2.status(task_id)
+ _status = task2.result.status(task_id)
  return {"status": _status, "task_id": task_id}
 
 
 @app.route("/serial/result/<string:task_id>")
-def serial_result(task_id: str) -> str:
+def serial_result(task_id: str) -> dict[str, Any]:
  """Define serial/result endpoint."""
  task1 = task_manager.get_task("serial")
- return task1.get_result(task_id)
+ result = None
+ for _ in range(10):
+ try:
+ # note: with no timeout
+ result = task1.result.get(task_id)
+ break
+ except Exception:
+ sleep(1)
+
+ if result is None:
+ return {"Error": "Result is not ready yet."}
+ return {"result": result[0]}
 
 
 @app.route("/parallel/result/<string:task_id>")
-def parallel_result(task_id: str) -> str:
+def parallel_result(task_id: str) -> dict[str, Any]:
  """Define parallel/result endpoint."""
  task2 = task_manager.get_task("parallel")
- return task2.get_result(task_id)
+
+ try:
+ # note: with timeout
+ result = task2.result.get(task_id, timeout=10)
+ except Exception:
+ return {"Error": "Result is not ready yet."}
+
+ return {"result": result[-1]}
 
 
 if __name__ == "__main__":

examples/redis_queue_between_tasks/run.sh

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+
+set -ex
+
+pushd ../../
+
+sugar build
+sugar ext restart --options -d
+sleep 5
+
+popd
+
+celery -A tasks.app worker --loglevel=debug &
+python app.py

examples/redis_queue_between_tasks/tasks/parallel.py

@@ -0,0 +1,66 @@
+"""My retsu tasks."""
+
+from __future__ import annotations
+
+from time import sleep
+
+import celery
+
+from retsu.celery import ParallelCeleryTask
+
+from .config import app, redis_client
+
+
+@app.task
+def task_parallel_a_plus_b(a: int, b: int, task_id: str) -> int: # type: ignore
+ """Define the task_parallel_a_plus_b."""
+ sleep(a + b)
+ print("running task_parallel_a_plus_b")
+ result = a + b
+ redis_client.set(f"parallel-result-a-plus-b-{task_id}", result)
+ return result
+
+
+@app.task
+def task_parallel_result_plus_10(task_id: str) -> int: # type: ignore
+ """Define the task_parallel_result_plus_10."""
+ print("running task_parallel_result_plus_10")
+ result = None
+ while result is None:
+ result = redis_client.get(f"parallel-result-a-plus-b-{task_id}")
+ sleep(1)
+
+ final_result = int(result) + 10
+ redis_client.set(f"parallel-result-plus-10-{task_id}", final_result)
+ return final_result
+
+
+@app.task
+def task_parallel_result_square(results, task_id: str) -> int: # type: ignore
+ """Define the task_parallel_result_square."""
+ print("running task_parallel_result_square")
+ result = None
+ while result is None:
+ result = redis_client.get(f"parallel-result-plus-10-{task_id}")
+ sleep(1)
+ return int(result) ** 2
+
+
+class MyParallelTask1(ParallelCeleryTask):
+ """MyParallelTask1."""
+
+ def request(self, a: int, b: int) -> str:
+ """Receive the request for processing."""
+ return super().request(a=a, b=b)
+
+ def get_chord_tasks(
+ self, a: int, b: int, task_id: str
+ ) -> list[celery.Signature]:
+ """Define the list of tasks for celery chord."""
+ return (
+ [
+ task_parallel_a_plus_b.s(a, b, task_id),
+ task_parallel_result_plus_10.s(task_id),
+ ],
+ task_parallel_result_square.s(task_id),
+ )