♻️ TasksManager uses Redis for task data (ITISFoundation#8131)

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

api/specs/web-server/_long_running_tasks_legacy.py

@@ -26,7 +26,7 @@
     name="list_tasks",
     description="Lists all long running tasks",
 )
-def list_tasks(): ...
+async def list_tasks(): ...
 
 
 @router.get(
@@ -35,7 +35,7 @@ def list_tasks(): ...
     name="get_task_status",
     description="Retrieves the status of a task",
 )
-def get_task_status(
+async def get_task_status(
     _path_params: Annotated[_PathParam, Depends()],
 ): ...
 
@@ -46,7 +46,7 @@ def get_task_status(
     description="Cancels and deletes a task",
     status_code=status.HTTP_204_NO_CONTENT,
 )
-def cancel_and_delete_task(
+async def cancel_and_delete_task(
     _path_params: Annotated[_PathParam, Depends()],
 ): ...
 
@@ -57,6 +57,6 @@ def cancel_and_delete_task(
     response_model=Any,
     description="Retrieves the result of a task",
 )
-def get_task_result(
+async def get_task_result(
     _path_params: Annotated[_PathParam, Depends()],
 ): ...

packages/celery-library/requirements/_test.in

@@ -11,6 +11,7 @@
 # testing
 coverage
 faker
+fakeredis[lua]
 httpx
 pint
 pytest

packages/celery-library/requirements/_test.txt

@@ -56,6 +56,8 @@ docker==7.1.0
     #   pytest-docker-tools
 faker==37.3.0
     # via -r requirements/_test.in
+fakeredis==2.30.3
+    # via -r requirements/_test.in
 flexcache==0.3
     # via pint
 flexparser==0.4
@@ -83,6 +85,8 @@ kombu==5.5.3
     #   -c requirements/_base.txt
     #   celery
     #   pytest-celery
+lupa==2.5
+    # via fakeredis
 packaging==25.0
     # via
     #   -c requirements/_base.txt
@@ -112,6 +116,10 @@ pygments==2.19.1
     # via
     #   -c requirements/_base.txt
     #   pytest
+pyjwt==2.9.0
+    # via
+    #   -c requirements/_base.txt
+    #   redis
 pytest==8.4.1
     # via
     #   -r requirements/_test.in
@@ -156,6 +164,11 @@ pyyaml==6.0.2
     #   -c requirements/../../../requirements/constraints.txt
     #   -c requirements/_base.txt
     #   -r requirements/_test.in
+redis==5.3.0
+    # via
+    #   -c requirements/../../../requirements/constraints.txt
+    #   -c requirements/_base.txt
+    #   fakeredis
 requests==2.32.4
     # via
     #   -c requirements/_base.txt
@@ -170,6 +183,8 @@ sniffio==1.3.1
     # via
     #   -c requirements/_base.txt
     #   anyio
+sortedcontainers==2.4.0
+    # via fakeredis
 tenacity==9.1.2
     # via
     #   -c requirements/_base.txt

packages/celery-library/src/celery_library/common.py

@@ -47,6 +47,8 @@ async def create_task_manager(
         ),
         client_name="celery_tasks",
     )
+    await redis_client_sdk.setup()
+    # GCR please address https://github.com/ITISFoundation/osparc-simcore/issues/8159
 
     return CeleryTaskManager(
         app,

packages/models-library/src/models_library/api_schemas_long_running_tasks/base.py

@@ -1,4 +1,5 @@
 import logging
+from collections.abc import Awaitable, Callable
 from typing import Annotated, TypeAlias
 
 from pydantic import BaseModel, Field, field_validator, validate_call
@@ -22,8 +23,16 @@ class TaskProgress(BaseModel):
     message: ProgressMessage = ""
     percent: ProgressPercent = 0.0
 
+    # used to propagate progress updates internally
+    _update_callback: Callable[["TaskProgress"], Awaitable[None]] | None = None
+
+    def set_update_callback(
+        self, callback: Callable[["TaskProgress"], Awaitable[None]]
+    ) -> None:
+        self._update_callback = callback
+
     @validate_call
-    def update(
+    async def update(
         self,
         *,
         message: ProgressMessage | None = None,
@@ -40,6 +49,16 @@ def update(
 
         _logger.debug("Progress update: %s", f"{self}")
 
+        if self._update_callback is not None:
+            try:
+                await self._update_callback(self)
+            except Exception as exc:  # pylint: disable=broad-exception-caught
+                _logger.warning(
+                    "Error while calling progress update callback: %s",
+                    exc,
+                    stack_info=True,
+                )
+
     @classmethod
     def create(cls, task_id: TaskId | None = None) -> "TaskProgress":
         return cls(task_id=task_id)

packages/pytest-simcore/src/pytest_simcore/redis_service.py

@@ -8,6 +8,7 @@
 
 import pytest
 import tenacity
+from fakeredis import FakeAsyncRedis
 from pytest_mock import MockerFixture
 from redis.asyncio import Redis, from_url
 from settings_library.basic_types import PortInt
@@ -121,3 +122,9 @@ def mock_redis_socket_timeout(mocker: MockerFixture) -> None:
     mocker.patch(
         "servicelib.redis._client.DEFAULT_SOCKET_TIMEOUT", timedelta(seconds=0.25)
     )
+
+
+@pytest.fixture
+async def use_in_memory_redis(mocker: MockerFixture) -> RedisSettings:
+    mocker.patch("redis.asyncio.from_url", FakeAsyncRedis)
+    return RedisSettings()

packages/service-library/requirements/_test.in

@@ -17,6 +17,7 @@ botocore
 coverage
 docker
 faker
+fakeredis[lua]
 flaky
 numpy
 openapi-spec-validator

packages/service-library/requirements/_test.txt

@@ -53,6 +53,8 @@ execnet==2.1.1
     # via pytest-xdist
 faker==36.1.1
     # via -r requirements/_test.in
+fakeredis==2.30.3
+    # via -r requirements/_test.in
 flaky==3.8.1
     # via -r requirements/_test.in
 frozenlist==1.5.0
@@ -109,6 +111,8 @@ jsonschema-specifications==2024.10.1
     #   openapi-schema-validator
 lazy-object-proxy==1.10.0
     # via openapi-spec-validator
+lupa==2.5
+    # via fakeredis
 multidict==6.1.0
     # via
     #   -c requirements/_aiohttp.txt
@@ -211,6 +215,11 @@ pyyaml==6.0.2
     #   -c requirements/_base.txt
     #   -c requirements/_fastapi.txt
     #   jsonschema-path
+redis==5.2.1
+    # via
+    #   -c requirements/../../../requirements/constraints.txt
+    #   -c requirements/_base.txt
+    #   fakeredis
 referencing==0.35.1
     # via
     #   -c requirements/../../../requirements/constraints.txt
@@ -245,6 +254,8 @@ sniffio==1.3.1
     #   -c requirements/_fastapi.txt
     #   anyio
     #   asgi-lifespan
+sortedcontainers==2.4.0
+    # via fakeredis
 sqlalchemy==1.4.54
     # via
     #   -c requirements/../../../requirements/constraints.txt

packages/service-library/src/servicelib/aiohttp/db_asyncpg_engine.py

@@ -8,7 +8,6 @@
 from typing import Final
 
 from aiohttp import web
-from servicelib.logging_utils import log_context
 from settings_library.postgres import PostgresSettings
 from simcore_postgres_database.utils_aiosqlalchemy import (  # type: ignore[import-not-found] # this on is unclear
     get_pg_engine_stateinfo,

packages/service-library/src/servicelib/aiohttp/long_running_tasks/_manager.py

@@ -1,9 +1,11 @@
 import datetime
 
 from aiohttp import web
+from settings_library.redis import RedisSettings
 
 from ...long_running_tasks.base_long_running_manager import BaseLongRunningManager
-from ...long_running_tasks.task import TaskContext, TasksManager
+from ...long_running_tasks.models import TaskContext
+from ...long_running_tasks.task import RedisNamespace, TasksManager
 from ._constants import APP_LONG_RUNNING_MANAGER_KEY
 from ._request import get_task_context
 
@@ -14,11 +16,15 @@ def __init__(
         app: web.Application,
         stale_task_check_interval: datetime.timedelta,
         stale_task_detect_timeout: datetime.timedelta,
+        redis_settings: RedisSettings,
+        redis_namespace: RedisNamespace,
     ):
         self._app = app
         self._tasks_manager = TasksManager(
             stale_task_check_interval=stale_task_check_interval,
             stale_task_detect_timeout=stale_task_detect_timeout,
+            redis_settings=redis_settings,
+            redis_namespace=redis_namespace,
         )
 
     @property