fodlder_id fix

Author: VaghinakDev

src/superannotate/lib/core/usecases/annotations.py

@@ -963,6 +963,7 @@ def execute(self):
                 self.reporter.warning_messages("Empty scores.")
         return self._response
 
+
 class DownloadAnnotations(BaseReportableUseCase):
     def __init__(
         self,
@@ -1053,96 +1054,100 @@ def coroutine_wrapper(coroutine):
         loop.close()
         return count
 
-    async def _download_big_annotation(self, item, export_path):
+    async def _download_big_annotation(self, item, export_path, folder_id):
         postfix = self.get_postfix()
-        response = await self._backend_client.download_big_annotation(
-            item = item,
+        await self._backend_client.download_big_annotation(
+            item=item,
             team_id=self._project.team_id,
             project_id=self._project.id,
-            folder_id=self._folder.uuid,
+            folder_id=folder_id,
             reporter=self.reporter,
             download_path=f"{export_path}{'/' + self._folder.name if not self._folder.is_root else ''}",
             postfix=postfix,
             callback=self._callback,
         )
 
-        return
-
-    async def download_big_annotations(self, queue_idx, export_path):
+    async def download_big_annotations(self, queue_idx, export_path, folder_id):
         while True:
             cur_queue = self._big_file_queues[queue_idx]
             item = await cur_queue.get()
             cur_queue.task_done()
             if item:
-                await self._download_big_annotation(item, export_path)
+                await self._download_big_annotation(item, export_path, folder_id)
             else:
                 cur_queue.put_nowait(None)
                 break
 
-    async def download_small_annotations(self, queue_idx, export_path):
-        max_chunk_size = 50000
-
+    async def download_small_annotations(self, queue_idx, export_path, folder_id):
         cur_queue = self._small_file_queues[queue_idx]
 
         items = []
-        i = 0
         item = ""
-
         postfix = self.get_postfix()
         while item is not None:
             item = await cur_queue.get()
             if item:
                 items.append(item)
-
-
         await self._backend_client.download_small_annotations(
-                team_id=self._project.team_id,
-                project_id=self._project.id,
-                folder_id=self._folder.uuid,
-                items=items,
-                reporter=self.reporter,
-                download_path=f"{export_path}{'/' + self._folder.name if not self._folder.is_root else ''}",
-                postfix=postfix,
-                callback=self._callback,
-            )
-
-    async def distribute_to_queues(self, item_names, sm_queue_id, l_queue_id, folder_id):
-
-        team_id=self._project.team_id
-        project_id=self._project.id
-
-        resp = self._backend_client.sort_items_by_size(item_names, team_id, project_id, folder_id)
+            team_id=self._project.team_id,
+            project_id=self._project.id,
+            folder_id=folder_id,
+            items=items,
+            reporter=self.reporter,
+            download_path=f"{export_path}{'/' + self._folder.name if not self._folder.is_root else ''}",
+            postfix=postfix,
+            callback=self._callback,
+        )
 
-        for item in resp['large']:
-            await self._big_file_queues[l_queue_id].put(item)
+    async def distribute_to_queues(
+        self, item_names, sm_queue_id, l_queue_id, folder_id
+    ):
+        try:
+            team_id = self._project.team_id
+            project_id = self._project.id
 
-        for item in resp['small']:
-            await self._small_file_queues[sm_queue_id].put(item['name'])
+            resp = self._backend_client.sort_items_by_size(
+                item_names, team_id, project_id, folder_id
+            )
 
+            for item in resp["large"]:
+                await self._big_file_queues[l_queue_id].put(item)
 
-        await self._big_file_queues[l_queue_id].put(None)
-        await self._small_file_queues[sm_queue_id].put(None)
+            for item in resp["small"]:
+                await self._small_file_queues[sm_queue_id].put(item["name"])
+        finally:
+            await self._big_file_queues[l_queue_id].put(None)
+            await self._small_file_queues[sm_queue_id].put(None)
 
     async def run_workers(self, item_names, folder_id, export_path):
         try:
             self._big_file_queues.append(asyncio.Queue())
             self._small_file_queues.append(asyncio.Queue())
             small_file_queue_idx = len(self._small_file_queues) - 1
             big_file_queue_idx = len(self._big_file_queues) - 1
-
             res = await asyncio.gather(
-                self.distribute_to_queues(item_names, small_file_queue_idx, big_file_queue_idx, folder_id),
-                self.download_big_annotations(big_file_queue_idx, export_path),
-                self.download_big_annotations(big_file_queue_idx, export_path),
-                self.download_big_annotations(big_file_queue_idx, export_path),
-                self.download_small_annotations(small_file_queue_idx, export_path),
-                return_exceptions=True
+                self.distribute_to_queues(
+                    item_names, small_file_queue_idx, big_file_queue_idx, folder_id
+                ),
+                self.download_big_annotations(
+                    big_file_queue_idx, export_path, folder_id
+                ),
+                self.download_big_annotations(
+                    big_file_queue_idx, export_path, folder_id
+                ),
+                self.download_big_annotations(
+                    big_file_queue_idx, export_path, folder_id
+                ),
+                self.download_small_annotations(
+                    small_file_queue_idx, export_path, folder_id
+                ),
+                return_exceptions=True,
             )
-
+            if any(res):
+                self.reporter.log_error(f"Error {str([i for i in res if i])}")
         except Exception as e:
             self.reporter.log_error(f"Error {str(e)}")
 
-
     def per_folder_execute(self, item_names, folder_id, export_path):
         asyncio.run(self.run_workers(item_names, folder_id, export_path))
 
@@ -1167,43 +1172,41 @@ def execute(self):
                     & Condition("project_id", self._project.id, EQ),
                 )
                 folders.append(self._folder)
-            postfix = self.get_postfix()
-
-            import nest_asyncio
-            import platform
-
-            if platform.system().lower() == "windows":
-                asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
-
-            nest_asyncio.apply()
 
             if not folders:
                 folders.append(self._folder)
-
-
             with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
                 futures = []
-
                 for folder in folders:
                     if not self._item_names:
                         condition = (
                             Condition("team_id", self._project.team_id, EQ)
                             & Condition("project_id", self._project.id, EQ)
                             & Condition("folder_id", folder.uuid, EQ)
                         )
-                        item_names = [item.name for item in self._images.get_all(condition)]
+                        item_names = [
+                            item.name for item in self._images.get_all(condition)
+                        ]
                     else:
                         item_names = self._item_names
 
                     new_export_path = export_path
-                    if folder.name != 'root':
-                        new_export_path +=  f'/{folder.name}'
-                    future = executor.submit(self.per_folder_execute, item_names, folder.uuid, new_export_path)
+                    if folder.name != "root":
+                        new_export_path += f"/{folder.name}"
+
+                    # TODO check
+                    if not item_names:
+                        continue
+                    future = executor.submit(
+                        self.per_folder_execute,
+                        item_names,
+                        folder.uuid,
+                        new_export_path,
+                    )
                     futures.append(future)
 
                 for future in concurrent.futures.as_completed(futures):
-                    pass
-
+                    print(future.result())
 
             self.reporter.stop_spinner()
             count = self.get_items_count(export_path)

src/superannotate/lib/infrastructure/services.py

@@ -8,14 +8,14 @@
 import time
 from contextlib import contextmanager
 from functools import lru_cache
+from pathlib import Path
 from typing import Any
 from typing import Callable
 from typing import Dict
 from typing import Iterable
 from typing import List
 from typing import Tuple
 from typing import Union
-from pathlib import Path
 from urllib.parse import urljoin
 
 import aiohttp
@@ -56,7 +56,7 @@ class BaseBackendService(SuperannotateServiceProvider):
     PAGINATE_BY = 100
     LIMIT = 100
     MAX_ITEMS_COUNT = 50 * 1000
-    ASSETS_PROVIDER_VERSION = 'v2'
+    ASSETS_PROVIDER_VERSION = "v2"
 
     """
     Base service class
@@ -80,16 +80,12 @@ def __init__(
         self._testing = testing
         self.get_session()
 
-
     @property
     def assets_provider_url(self):
         if self.api_url != constants.BACKEND_URL:
             return f"https://assets-provider.devsuperannotate.com/api/{self.ASSETS_PROVIDER_VERSION}/"
         return f"https://assets-provider.superannotate.com/api/{self.ASSETS_PROVIDER_VERSION}/"
 
-
-
-
     @lru_cache(maxsize=32)
     def _get_session(self, thread_id, ttl=None):  # noqa
         del ttl
@@ -1165,69 +1161,66 @@ async def download_big_annotation(
         download_path: str,
         postfix: str,
         item: int,
-        callback: Callable = None
+        callback: Callable = None,
     ):
-        item_id = item['id']
-        item_name = item['name']
+        item_id = item["id"]
+        item_name = item["name"]
         query_params = {
-            "team_id" : team_id,
+            "team_id": team_id,
             "project_id": project_id,
             "annotation_type": "MAIN",
-            "version":"V1.00"
+            "version": "V1.00",
         }
 
-        url = urljoin(self.assets_provider_url, self.URL_DOWNLOAD_LARGE_ANNOTATION.format(item_id=item_id))
+        url = urljoin(
+            self.assets_provider_url,
+            self.URL_DOWNLOAD_LARGE_ANNOTATION.format(item_id=item_id),
+        )
 
         sync_params = {
             "team_id": team_id,
             "project_id": project_id,
             "desired_transform_version": "export",
-            "desired_version":"V1.00",
+            "desired_version": "V1.00",
             "current_transform_version": "V1.00",
-            "current_source":"main",
-            "desired_source":"secondary"
+            "current_source": "main",
+            "desired_source": "secondary",
         }
 
-        sync_url = urljoin(self.assets_provider_url, self.URL_SYNC_LARGE_ANNOTATION.format(item_id=item_id))
-
-        res = self._request(
-            url = sync_url,
-            params = sync_params,
-            method="POST"
+        sync_url = urljoin(
+            self.assets_provider_url,
+            self.URL_SYNC_LARGE_ANNOTATION.format(item_id=item_id),
         )
 
-        sync_params.pop('current_source')
-        sync_params.pop('desired_source')
+        res = self._request(url=sync_url, params=sync_params, method="POST")
 
+        sync_params.pop("current_source")
+        sync_params.pop("desired_source")
 
         synced = False
 
-        sync_status_url = urljoin(self.assets_provider_url, self.URL_SYNC_LARGE_ANNOTATION_STATUS.format(item_id=item_id))
-        while synced != 'SUCCESS':
+        sync_status_url = urljoin(
+            self.assets_provider_url,
+            self.URL_SYNC_LARGE_ANNOTATION_STATUS.format(item_id=item_id),
+        )
+        while synced != "SUCCESS":
             synced = self._request(
-                url = sync_status_url,
-                params = sync_params,
-                method = "GET"
-            ).json()['status']
+                url=sync_status_url, params=sync_params, method="GET"
+            ).json()["status"]
             await asyncio.sleep(1)
 
-
         async with aiohttp.ClientSession(
-            connector=aiohttp.TCPConnector(ssl=False), headers = self.default_headers,
+            connector=aiohttp.TCPConnector(ssl=False),
+            headers=self.default_headers,
         ) as session:
-
-            start_response = await session.post(
-                url,
-                params = query_params
-            )
+            start_response = await session.post(url, params=query_params)
             res = await start_response.json()
             Path(download_path).mkdir(exist_ok=True, parents=True)
 
             dest_path = Path(download_path) / (item_name + postfix)
-            with open(dest_path, 'w') as fp:
+            with open(dest_path, "w") as fp:
                 json.dump(res, fp)
 
-
     async def download_small_annotations(
         self,
         project_id: int,
@@ -1245,7 +1238,6 @@ async def download_small_annotations(
             "project_id": project_id,
             "folder_id": folder_id,
         }
-
         handler = StreamedAnnotations(
             headers=self.default_headers,
             reporter=reporter,
@@ -1596,20 +1588,10 @@ def sort_items_by_size(self, item_names, team_id, project_id, folder_id):
             "folder_id": folder_id,
         }
 
-        url = urljoin(
-            self.assets_provider_url, self.URL_CLASSIFY_ITEM_SIZE
-        )
-
-        body = {
-            "item_names": item_names,
-            "folder_id": folder_id
-        }
-
-
-        return self._request(
-            url = url,
-            method = "POST",
-            params = query_params,
-            data = body
-        ).json()
+        url = urljoin(self.assets_provider_url, self.URL_CLASSIFY_ITEM_SIZE)
 
+        body = {"item_names": item_names, "folder_id": folder_id}
+        response = self._request(url=url, method="POST", params=query_params, data=body)
+        if not response.ok:
+            raise AppException(response.json().get("errors", "Undefined"))
+        return response.json()