feat: draft async get assets

Author: dfguerrerom

sepal_ui/scripts/gee.py

@@ -1,5 +1,7 @@
 """All the heleper methods to interface Google Earthengine with sepal-ui."""
 
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
 import time
 from pathlib import Path
 from typing import List, Union
@@ -83,8 +85,59 @@ def is_running(task_descripsion: str) -> ee.batch.Task:
     return current_task
 
 
+async def list_assets_concurrent(folders):
+    with ThreadPoolExecutor() as executor:
+        loop = asyncio.get_running_loop()
+        tasks = [
+            loop.run_in_executor(executor, ee.data.listAssets, {"parent": folder})
+            for folder in folders
+        ]
+        results = await asyncio.gather(*tasks)
+        return results
+
+
+async def get_assets_async_concurrent(folder: str) -> list:
+    folder_queue = asyncio.Queue()
+    await folder_queue.put(folder)
+    asset_list = []
+
+    while not folder_queue.empty():
+        current_folders = [await folder_queue.get() for _ in range(folder_queue.qsize())]
+        assets_groups = await list_assets_concurrent(current_folders)
+
+        for assets in assets_groups:
+            for asset in assets.get("assets", []):
+                asset_list.append({"type": asset["type"], "name": asset["name"], "id": asset["id"]})
+                if asset["type"] == "FOLDER":
+                    await folder_queue.put(asset["name"])
+
+    return asset_list
+
+
 @sd.need_ee
-def get_assets(folder: Union[str, Path] = "") -> List[dict]:
+def get_assets(folder: Union[str, Path] = "", async_=True) -> List[dict]:
+    """Get all the assets from the parameter folder. every nested asset will be displayed.
+
+    Args:
+
+        folder: the initial GEE folder
+        async_: whether or not the function should be executed asynchronously
+
+    Returns:
+        the asset list. each asset is a dict with 3 keys: 'type', 'name' and 'id'
+
+    """
+
+    folder = str(folder) or f"projects/{ee.data._cloud_api_user_project}/assets/"
+
+    if async_:
+        return asyncio.run(get_assets_async_concurrent(folder))
+
+    return get_assets_sync(folder)
+
+
+@sd.need_ee
+def get_assets_sync(folder: Union[str, Path] = "") -> List[dict]:
     """Get all the assets from the parameter folder. every nested asset will be displayed.
 
     Args:
@@ -95,7 +148,6 @@ def get_assets(folder: Union[str, Path] = "") -> List[dict]:
     """
     # set the folder and init the list
     asset_list = []
-    folder = str(folder) or f"projects/{ee.data._cloud_api_user_project}/assets/"
 
     def _recursive_get(folder, asset_list):
 
@@ -151,6 +203,7 @@ def delete_assets(asset_id: str, dry_run: bool = True) -> None:
         asset_id: the Id of the asset or a folder
         dry_run: whether or not a dry run should be launched. dry run will only display the files name without deleting them.
     """
+
     # define the action to execute for each asset based on the dry run mode
     def delete(id: str) -> None:
         if dry_run is True: