fix: no-stream

Author: giancarloromeo

packages/celery-library/src/celery_library/backends/redis.py

@@ -1,7 +1,5 @@
 import contextlib
-import json
 import logging
-from collections.abc import AsyncIterable
 from datetime import timedelta
 from typing import TYPE_CHECKING, Final
 
@@ -127,36 +125,6 @@ async def list_tasks(self, task_filter: TaskFilter) -> list[Task]:
     async def remove_task(self, task_id: TaskID) -> None:
         await self._redis_client_sdk.redis.delete(_build_key(task_id))
 
-    async def append_task_result(self, task_id: TaskID, data: dict) -> None:
-        stream_key = f"task:{task_id}"
-        await self._redis_client_sdk.redis.xadd(
-            stream_key,
-            data,
-        )
-
-    async def stream_task_result(
-        self, task_id: str, last_id: str = "0-0"
-    ) -> AsyncIterable[dict]:
-        stream_key = f"task:{task_id}"
-        while True:
-            result = await self._redis_client_sdk.redis.xread(
-                {stream_key: last_id}, block=5000
-            )
-            if not result:
-                continue
-
-            for _, entries in result:
-                for entry_id, fields in entries:
-                    last_id = entry_id
-                    data = {
-                        k: json.loads(v) if k == "data" else v
-                        for k, v in fields.items()
-                    }
-                    yield data
-
-                    if data.get("type") == "done":
-                        return
-
     async def set_task_progress(self, task_id: TaskID, report: ProgressReport) -> None:
         await handle_redis_returns_union_types(
             self._redis_client_sdk.redis.hset(

services/storage/src/simcore_service_storage/api/_worker_tasks/_simcore_s3.py

@@ -151,24 +151,18 @@ async def search(
 
         assert isinstance(dsm, SimcoreS3DataManager)  # nosec
 
-        pages: list[SearchResult] = []
-        async for page in dsm.search(
-            user_id=user_id,
-            project_id=project_id,
-            name_pattern=name_pattern,
-        ):
-            # TODO: publish temporary result
-            pages.extend(
-                [
-                    SearchResult(
-                        name=item.file_name,
-                        project_id=item.project_id,
-                        created_at=item.created_at,
-                        modified_at=item.last_modified,
-                        is_directory=item.is_directory,
-                    )
-                    for item in page
-                ]
+        return [
+            SearchResult(
+                name=item.file_name,
+                project_id=item.project_id,
+                created_at=item.created_at,
+                modified_at=item.last_modified,
+                is_directory=item.is_directory,
             )
-
-        return pages
+            async for page in dsm.search(
+                user_id=user_id,
+                project_id=project_id,
+                name_pattern=name_pattern,
+            )
+            for item in page
+        ]

services/storage/src/simcore_service_storage/simcore_s3_dsm.py

@@ -1061,7 +1061,7 @@ async def search(
 
         Args:
             user_id: The user requesting the search
-            filename_pattern: Wildcard pattern for filename matching (e.g., "*.txt", "test_*.json")
+            name_pattern: Wildcard pattern for filename matching (e.g., "*.txt", "test_*.json")
             project_id: Optional project ID to limit search to specific project
             items_per_page: Number of items to return per page