hacktoberfest - Added a get all results function

covalent/_results_manager/results_manager.py

@@ -20,7 +20,7 @@
 import contextlib
 import os
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Union
 
 from furl import furl
 from requests.adapters import HTTPAdapter
@@ -93,21 +93,21 @@
         FileNotFoundError: If the result file is not found.
     """
 
     if results_dir is None:
         results_dir = os.environ.get("COVALENT_DATA_DIR") or get_config("dispatcher.results_dir")
 
     import shutil
 
     result_folder_path = os.path.join(results_dir, f"{dispatch_id}")
 
     if os.path.exists(result_folder_path):
         shutil.rmtree(result_folder_path, ignore_errors=True)
 
     with contextlib.suppress(OSError):
         os.rmdir(results_dir)
 
     if remove_parent_directory:
         shutil.rmtree(results_dir, ignore_errors=True)
 
 
 def cancel(dispatch_id: str, task_ids: List[int] = None, dispatcher_addr: str = None) -> str:
@@ -127,7 +127,7 @@
         dispatcher_addr = format_server_url()
 
     if task_ids is None:
         task_ids = []
 
     api_client = CovalentAPIClient(dispatcher_addr)
     endpoint = f"/api/v2/dispatches/{dispatch_id}/status"
@@ -450,6 +450,74 @@
 
     return rm.result_object
 
+def get_all_results(
+    dispatch_ids: Optional[List[str]] = None,
+    statuses: Optional[List[str]] = None,
+    started_before: Optional[str] = None,
+    started_after: Optional[str] = None,
+    completed_before: Optional[str] = None,
+    completed_after: Optional[str] = None,
+) -> List[Dict[str, Union[str, "Result"]]]:
+    """
+    Retrieve all results.
+
+    Parameters:
+        dispatch_ids: Optional[List[str]]
+            A list of dispatch IDs to filter results by.
+
+        statuses: Optional[List[str]]
+            A list of statuses to filter results by.
+
+        started_before: Optional[str]
+            Only return results started before this timestamp.
+
+        started_after: Optional[str]
+            Only return results started after this timestamp.
+
+        completed_before: Optional[str]
+            Only return results completed before this timestamp.
+
+        completed_after: Optional[str]
+            Only return results completed after this timestamp.
+
+    Returns:
+        List[Dict[str, Union[str, "Result"]]]: A list of dictionaries, where each dictionary contains a dispatch ID and its corresponding result.
+    """
+    config_results_dir = ct.get_config()["dispatcher"]["results_dir"]
+    all_dispatch_ids = os.listdir(config_results_dir)
+
+    # Populate the docs
+    docs = []
+    for d_id in all_dispatch_ids:
+        if dispatch_ids and d_id not in dispatch_ids:
+            continue
+
+        try:
+            result_obj = ct.get_result(d_id)
+        except MissingLatticeRecordError:
+            continue
+
+        if not result_obj:
+            continue
+
+        if statuses and result_obj.status not in statuses:
+            continue
+
+        if started_before and result_obj.started > started_before:
+            continue
+
+        if started_after and result_obj.started < started_after:
+            continue
+
+        if completed_before and result_obj.completed > completed_before:
+            continue
+
+        if completed_after and result_obj.completed < completed_after:
+            continue
+
+        docs.append({"dispatch_id": d_id, "result": result_obj})
+
+    return docs
 
 def get_result(
     dispatch_id: str,