feat(tasks): improved retrieval

Author: ChrisCoder9000

README.md

@@ -1,5 +1,5 @@
 <p align="center">
-  <img src="https://img.shields.io/badge/version-2.0.11--dev-blue?style=for-the-badge" alt="Version"/>
+  <img src="https://img.shields.io/badge/version-2.1.0--dev-blue?style=for-the-badge" alt="Version"/>
   <img src="https://img.shields.io/badge/python-3.11+-green?style=for-the-badge&logo=python&logoColor=white" alt="Python"/>
   <img src="https://img.shields.io/badge/license-AGPLv3%20%2B%20Commons%20Clause-purple?style=for-the-badge" alt="License"/>
 </p>

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "brainapi2"
-version = "2.0.11-dev"
+version = "2.1.0-dev"
 description = "Version 2.x.x of the BrainAPI memory layer."
 authors = [
     {name = "Christian",email = "alch.infoemail@gmail.com"}

src/adapters/cache.py

@@ -49,3 +49,9 @@ def delete(self, key: str, brain_id: str = "default") -> bool:
         Delete a value from the cache.
         """
         return self.cache.delete(key, brain_id)
+
+    def get_task_keys(self, brain_id: str = "default") -> list[str]:
+        """
+        Get all task keys for a given brain_id.
+        """
+        return self.cache.get_task_keys(brain_id)

src/adapters/interfaces/cache.py

@@ -47,3 +47,10 @@ def delete(self, key: str, brain_id: str) -> bool:
         Delete a value from the cache.
         """
         raise NotImplementedError("delete method not implemented")
+
+    @abstractmethod
+    def get_task_keys(self, brain_id: str) -> list[str]:
+        """
+        Get all task keys for a given brain_id.
+        """
+        raise NotImplementedError("get_task_keys method not implemented")

src/lib/redis/client.py

@@ -57,6 +57,8 @@ def set(
         Set a value in the cache with an expiration time.
         """
         prefixed_key = self._get_key(key, brain_id)
+        if key.startswith("task:"):
+            self.client.sadd(f"{brain_id}:_task_index", key)
         return self.client.set(
             prefixed_key, value, **({"ex": expires_in} if expires_in else {})
         )
@@ -66,7 +68,16 @@ def delete(self, key: str, brain_id: str) -> bool:
         Delete a value from the cache.
         """
         prefixed_key = self._get_key(key, brain_id)
+        if key.startswith("task:"):
+            self.client.srem(f"{brain_id}:_task_index", key)
         return self.client.delete(prefixed_key)
 
+    def get_task_keys(self, brain_id: str) -> list[str]:
+        """
+        Get all task keys for a given brain_id.
+        """
+        keys = self.client.smembers(f"{brain_id}:_task_index")
+        return [k.decode("utf-8") if isinstance(k, bytes) else k for k in keys]
+
 
 _redis_client = RedisClient()

src/services/api/routes/tasks.py

@@ -7,13 +7,37 @@
 Modified By: the developer formerly known as Christian Nonis at <alch.infoemail@gmail.com>
 -----
 """
+
 import json
 from src.utils.logging import log
 from fastapi import APIRouter
 from src.services.kg_agent.main import cache_adapter
 
 tasks_router = APIRouter(prefix="/tasks", tags=["tasks"])
 
+
+@tasks_router.get("/")
+async def get_tasks(brain_id: str = "default"):
+    try:
+        tasks = cache_adapter.get_task_keys(brain_id)
+        results = []
+        for task in tasks:
+            str_result = cache_adapter.get(task, brain_id=brain_id)
+            if str_result is None:
+                continue
+            result = json.loads(str_result)
+            results.append(
+                {"id": task.split(":")[-1], "status": result.get("status", "unknown")}
+            )
+        return {"tasks": results}
+    except Exception as e:
+        log(f"Error in get_tasks: {type(e).__name__}: {str(e)}")
+        return {
+            "status": "error",
+            "result": {"error": str(e), "error_type": type(e).__name__},
+        }
+
+
 @tasks_router.get("/{task_id}")
 async def get_task(task_id: str, brain_id: str = "default"):
     """
@@ -25,7 +49,7 @@ async def get_task(task_id: str, brain_id: str = "default"):
             return {
                 "task_id": task_id,
                 "status": "pending",
-                "result": {"message": "Task is still processing or not found"}
+                "result": {"message": "Task is still processing or not found"},
             }
         if isinstance(str_result, bytes):
             result = json.loads(str_result.decode("utf-8"))
@@ -34,15 +58,12 @@ async def get_task(task_id: str, brain_id: str = "default"):
         return {
             "task_id": task_id,
             "status": result.get("status", "unknown"),
-            "result": result
+            "result": result,
         }
     except Exception as e:
         log(f"Error in get_task: {type(e).__name__}: {str(e)}")
         return {
             "task_id": task_id,
             "status": "error",
-            "result": {
-                "error": str(e),
-                "error_type": type(e).__name__
-            }
-        }
+            "result": {"error": str(e), "error_type": type(e).__name__},
+        }