Merge pull request #722 from superannotateai/FRIDAY-3252

Friday 3252

Author: VaghinakDev

src/superannotate/lib/app/interface/sdk_interface.py

@@ -2663,23 +2663,23 @@ def search_items(
 
     def list_items(
         self,
-        project: Union[int, str],
-        folder: Optional[Union[int, str]] = None,
+        project: Union[NotEmptyStr, int],
+        folder: Optional[Union[NotEmptyStr, int]] = None,
         *,
         include: List[Literal["custom_metadata"]] = None,
         **filters,
     ):
         """
         Search items by filtering criteria.
 
-        :param project: The project name, project ID, or folder path (e.g., "project1/folder1") to search within.
+        :param project: The project name, project ID, or folder path (e.g., "project1") to search within.
                         This can refer to the root of the project or a specific subfolder.
-        :type project: Union[int, str]
+        :type project: Union[NotEmptyStr, int]
 
         :param folder: The folder name or ID to search within. If None, the search will be done in the root folder of
                        the project. If both “project” and “folder” specify folders, the “project”
                        value will take priority.
-        :type folder: Union[int, str], optional
+        :type folder: Union[NotEmptyStr, int], optional
 
         :param include: Specifies additional fields to include in the response.
 
@@ -2694,8 +2694,8 @@ def list_items(
 
 
                 Supported operations:
-                    - __ne: Value is in the list.
-                    - __in: Value is not equal.
+                    - __ne: Value is not equal.
+                    - __in: Value is in the list.
                     - __notin: Value is not in the list.
                     - __contains: Value has the substring.
                     - __starts: Value starts with the prefix.
@@ -2710,6 +2710,9 @@ def list_items(
                 - name__contains: str
                 - name__starts: str
                 - name__ends: str
+                - annotation_status: str
+                - annotation_status__in: list[str]
+                - annotation_status__ne: list[str]
                 - approval_status: Literal["Approved", "Disapproved", None]
                 - assignments__user_id: str
                 - assignments__user_id__ne: str

src/superannotate/lib/infrastructure/controller.py

@@ -60,6 +60,7 @@ class ItemFilters(TypedDict, total=False):
     name__ends: Optional[str]
     annotation_status: Optional[str]
     annotation_status__in: Optional[List[str]]
+    annotation_status__ne: Optional[List[str]]
     approval_status: Optional[str]
     approval_status__in: Optional[List[str]]
     approval_status__ne: Optional[str]

src/superannotate/lib/infrastructure/stream_data_handler.py

@@ -1,5 +1,6 @@
 import copy
 import json
+import logging
 import os
 import typing
 from typing import Callable
@@ -13,6 +14,8 @@
     total=_seconds, sock_connect=_seconds, sock_read=_seconds
 )
 
+logger = logging.getLogger("sa")
+
 
 class StreamedAnnotations:
     DELIMITER = b"\\n;)\\n"
@@ -46,21 +49,32 @@ async def fetch(
         data: dict = None,
         params: dict = None,
     ):
-        kwargs = {"params": params, "json": {}}
-        if "folder_id" in kwargs["params"]:
-            kwargs["json"] = {"folder_id": kwargs["params"].pop("folder_id")}
+        kwargs = {"params": params, "json": data}
         if data:
             kwargs["json"].update(data)
         response = await session.request(method, url, **kwargs, timeout=TIMEOUT)  # noqa
-        buffer = b""
+        if not response.ok:
+            logger.error(response.text)
+        buffer = ""
+        line_groups = b""
+        decoder = json.JSONDecoder()
         async for line in response.content.iter_any():
-            slices = (buffer + line).split(self.DELIMITER)
-            for _slice in slices[:-1]:
-                yield self.get_json(_slice)
-            buffer = slices[-1]
-        if buffer:
-            yield self.get_json(buffer)
-            self._reporter.update_progress()
+            line_groups += line
+            try:
+                buffer += line_groups.decode("utf-8")
+                line_groups = b""
+            except UnicodeDecodeError:
+                continue
+            while buffer:
+                try:
+                    json_obj, index = decoder.raw_decode(buffer)
+                    yield json_obj
+                    buffer = buffer[index + len(self.DELIMITER) :].lstrip()
+                except json.decoder.JSONDecodeError as e:
+                    logger.debug(
+                        f"Failed to parse buffer, buffer_len: {len(buffer)}// buffer_end: ...{buffer[-100:]}, error: {e}"
+                    )
+                    break
 
     async def list_annotations(
         self,