Clear_global_keys

Author: Oleg Trygub

labelbox/client.py

@@ -1229,6 +1229,110 @@ def _format_failed_rows(rows: List[str],
                 )
             time.sleep(sleep_time)
 
+    def clear_global_keys(
+            self,
+            global_keys: List[str],
+            timeout_seconds=60) -> Dict[str, Union[str, List[Any]]]:
+        """
+        Gets data row ids for a list of global keys.
+
+        Args:
+            A list of global keys
+        Returns:
+            Dictionary containing 'status', 'results' and 'errors'.
+
+            'Status' contains the outcome of this job. It can be one of
+            'Success', 'Partial Success', or 'Failure'.
+
+            'Results' contains a list of data row ids successfully fetchced. It may
+            not necessarily contain all data rows requested.
+
+            'Errors' contains a list of global_keys that could not be fetched, along
+            with the failure reason
+        Examples:
+            >>> job_result = client.get_data_row_ids_for_global_keys(["key1","key2"])
+            >>> print(job_result['status'])
+            Partial Success
+            >>> print(job_result['results'])
+            ['cl7tv9wry00hlka6gai588ozv', 'cl7tv9wxg00hpka6gf8sh81bj']
+            >>> print(job_result['errors'])
+            [{'global_key': 'asdf', 'error': 'Data Row not found'}]
+        """
+
+        def _format_global_key_er(rows: List[str],
+                                error_msg: str) -> List[Dict[str, str]]:
+            return [{'global_key': r, 'error': error_msg} for r in rows]
+
+        # Start get data rows for global keys job
+        query_str = """mutation clearGlobalKeysPyApi($globalKeys: [ID!]!) {
+            clearGlobalKeys(where: {ids: $globalKeys}) { jobId}}
+            """
+        params = {"globalKeys": global_keys}
+        clear_global_keys_job = self.execute(query_str, params)
+
+        # Query string for retrieving job status and result, if job is done
+        result_query_str = """query clearGlobalKeysResultPyApi($jobId: ID!) {
+            clearGlobalKeysResult(jobId: {id: $jobId}) { data { 
+                clearedGlobalKeys
+                failedToClearGlobalKeys
+                notFoundGlobalKeys
+                accessDeniedGlobalKeys
+                } jobStatus}}
+            """
+        result_params = {
+            "jobId":
+                clear_global_keys_job["clearGlobalKeys"]["jobId"]
+        }
+
+        def _format_failed_rows(rows: List[str],
+                                error_msg: str) -> List[Dict[str, str]]:
+            return [{'global_key': r, 'error': error_msg} for r in rows]
+
+        # Poll job status until finished, then retrieve results
+        sleep_time = 2
+        start_time = time.time()
+        while True:
+            res = self.execute(result_query_str, result_params)
+            if res["clearGlobalKeysResult"]['jobStatus'] == "COMPLETE":
+                data = res["clearGlobalKeysResult"]['data']
+                results, errors = [], []
+                results.extend(data['clearedGlobalKeys'])
+                errors.extend(
+                    _format_failed_rows(data['failedToClearGlobalKeys'],
+                                        "Clearing global key failed"))
+                errors.extend(
+                    _format_failed_rows(data['notFoundGlobalKeys'],
+                                        "Failed to find data row matching provided global key"))
+                errors.extend(
+                    _format_failed_rows(data['accessDeniedGlobalKeys'],
+                                        "Denied access to modify data row matching provided global key"))
+
+                # Invalid results may contain empty string, so we must filter
+                # them prior to checking for PARTIAL_SUCCESS
+                filtered_results = list(filter(lambda r: r != '', results))
+                if not errors:
+                    status = CollectionJobStatus.SUCCESS.value
+                elif errors and len(filtered_results) > 0:
+                    status = CollectionJobStatus.PARTIAL_SUCCESS.value
+                else:
+                    status = CollectionJobStatus.FAILURE.value
+
+                if errors:
+                    logger.warning(
+                        "There are errors present. Please look at 'errors' in the returned dict for more details"
+                    )
+
+                return {"status": status, "results": results, "errors": errors}
+            elif res["clearGlobalKeysResult"]['jobStatus'] == "FAILED":
+                raise labelbox.exceptions.LabelboxError(
+                    "Job clearGlobalKeys failed.")
+            current_time = time.time()
+            if current_time - start_time > timeout_seconds:
+                raise labelbox.exceptions.TimeoutError(
+                    "Timed out waiting for clear_global_keys job to complete."
+                )
+            time.sleep(sleep_time)
+
     def get_catalog_slice(self, slice_id) -> CatalogSlice:
         """
         Fetches a Catalog Slice by ID.

labelbox_dev/fetching_iterator.py

@@ -0,0 +1,69 @@
+from collections import deque
+
+TASK_ID_KEY = 'jobID'
+TASK_STATUS_KEY = 'jobStatus'
+TOTAL_PAGES_KEY = 'totalPage'
+PAGE_NUMBER_KEY = 'pageNumber'
+DATA_KEY = 'data'
+DEFAULT_QUERY = ''
+DEFAULT_QUERY_NAME = ''
+
+class FetchingIterator:
+    def __init__(self, 
+                object_class,
+                query, 
+                params, 
+                prefetched_page_json = None, 
+                total_pages = None):
+        self.queue = deque()
+        self.object_class = object_class
+        self.query = query
+        self.params = params
+        self.iteration_ended = False
+        self.page_number = 0
+        self.total_pages = 0
+
+        self.__fetch_next_page(prefetched_page_json, total_pages)
+
+
+    def __len__(self):
+        return len(self._queue)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if not self._queue and not self._fetch_next_page():
+            raise StopIteration()
+
+        return self.queue.popleft()
+
+    def __instatiate_typed_entities(self, json_entities):
+        entities = []
+        for json_entity in json_entities:
+            entities.append(self.object_class(json_entity))
+        return entities
+    
+    def __parse_response(self, response):
+        total_pages = response.get(TOTAL_PAGES_KEY, 0)
+        page_number = response.get(PAGE_NUMBER_KEY, 0)
+        json_entities = response.get(DATA_KEY, {})
+        return json_entities, page_number, total_pages
+    
+    def __fetch_next_page(self, prefetched_page, total_pages):
+        if self.iteration_ended:
+            return False
+        
+        if not prefetched_page:
+            page_response = self.session.execute(self.query, self.params)
+            json_entities, self.page_number, self.total_pages = self.__parse_response(page_response)
+        else:
+            json_entities, self.page_number, self.total_pages = self.__parse_response(prefetched_page)
+       
+        entities = self.__instatiate_typed_entities(json_entities)
+        self.queue.append(entities)
+        
+        if self.page_number == total_pages:
+            self.iteration_ended = True
+
+        return True

labelbox_dev/globalkeys.py

@@ -0,0 +1,75 @@
+from labelbox_dev import entEnity
+
+from collections import namedtuple
+
+from pydantic import NoneStr
+sys.path.append("/Users/olegtrygub/src/labelbox-python")
+import labelbox
+
+from labelbox.client import Client
+from labelbox.utils import camel_case
+
+from typing import Optional
+
+from time import time
+
+client = labelbox.Client(
+                api_key="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VySWQiOiJjbDh1dHpxMm8wMDBieHBwMTR0a2o4ZWJzIiwib3JnYW5pemF0aW9uSWQiOiJjbDh1dHpxMXAwMDBheHBwMWNrdHhod3p0IiwiYXBpS2V5SWQiOiJjbDh1dTB4b2YwMDBleHBwMWQwcmMweDhsIiwic2VjcmV0IjoiMmExNDQ4NGZmZmIzMjQyNzk0ZTNlNmIwOTRkMDY3OWYiLCJpYXQiOjE2NjQ5MjYxMjIsImV4cCI6MjI5NjA3ODEyMn0.cqtxIEKVY1GPAQjyqQ8VrbMGkUywx-IwC9BGkCi8ahU",
+                endpoint = 'http://localhost:8080/graphql', 
+                app_url="localhost:3000"
+            )
+
+
+class BaseObject:
+    def create_from_json(self):
+        pass
+
+
+class DataRowInfo(Entity):
+    def __init__(self):
+        self.data_row_id = None
+        self.globalkey = None
+
+class DataRowAssignment(Entity):
+    pass
+
+class BaseError(BaseObject):
+    def __init__(self, message):
+        self.message = message
+
+
+
+
+TASK_ID_KEY = 'jobID'
+TASK_STATUS_KEY = 'jobStatus'
+TOTAL_PAGES_KEY = 'totalPage'
+PAGE_NUMBER_KEY = 'pageNumber'
+DATA_KEY = 'data'
+DEFAULT_QUERY = ''
+DEFAULT_QUERY_NAME = ''
+
+
+DEFAULT_PAGE_SCHEMA = {}
+
+
+
+
+
+def get_data_row_ids_for_globaleys(globalkeys):
+    pass
+
+
+def assign_global_keys_to_data_rows(global_key_to_data_row_inputs: List[Dict[str, str]], timeout_seconds=60) -> Dict[str, Union[str, List[Any]]]:
+    pass
+
+if __name__ == "__main__":
+
+
+
+    PATTERN = "gs://labelbox-datasets/coco_dataset/train2017/00000000{}.jpg-1"
+    gks = [PATTERN.format(num ) for num in range(1006, 1300)]
+    dr_gk_ids = client.get_data_row_ids_for_global_keys(gks, timeout_seconds=4)
+    print(dr_gk_ids.keys())
+
+
+

labelbox_dev/task.py

@@ -0,0 +1,91 @@
+from typing import TYPE_CHECKING, List, Dict, Any, Union, Optional
+from time import time
+from datetime import datetime
+from enum import Enum
+
+class TaskStatus(Enum):
+    SUBMITTED = 0
+    COMPLETE = 1
+    RUNNING = 2
+    FAILED = 3
+    RETRYING = 3
+
+class TaskType(Enum):
+    BEST_EFFORT = 0
+    TRANSACTION = 1
+
+class BaseTask:
+    def __init__(self, task_id, object_class, metadata_query, query_name, created_at, expires_at = None):
+        self._session = client
+
+        self._metadata_query = metadata_query
+        self._query_name = query_name
+        self._task_id = task_id
+        self.object_class = object_class
+
+        self._status = None
+        self._created_at = created_at
+        self._expires_at = expires_at
+
+        
+    def status(self) -> Optional[TaskStatus]:
+        # fetch if stale 
+        if self._status is None or self.status == TaskStatus.RUNNING:
+            self.__fetch_task_metadata()
+        
+        return self._status
+ 
+    def wait_till_done(self):
+        pass
+ 
+    def created_at(self) -> time:
+        return self._created_at
+ 
+    def expires_at(self) -> Optional[time]:
+        pass
+
+    def __fetch_task_metadata(self):
+        response = self._session.execute(self._metadata_query, {TASK_ID_KEY: self.task_id})
+        self.task_status = response[TASK_STATUS_KEY]
+
+
+class Task(BaseTask): 
+    def error(self) -> Optional[BaseError]:
+        pass
+ 
+    def result(self) -> Optional[BaseObject]:
+        pass
+
+class BulkTask(BaseTask):
+    def __init__(
+        self, 
+        task_id, 
+        object_class,
+        task_type = TaskType.TRANSACTION,
+        query = DEFAULT_QUERY, 
+        query_name = DEFAULT_QUERY_NAME, 
+        created_at = datetime.now()
+    ):
+        super(BaseTask, self).__init__(task_id, query, query_name, created_at, object_class)
+        self.results_iterator = None
+        self.errors_iterator = None
+
+    def results(self) -> Optional[FetchingIterator]:
+        if self.status() == TaskStatus.COMPLETE:
+            if self.results_iterator:
+                return self.results_iterator
+
+            self.results_iterator = FetchingIterator()
+        return None
+ 
+    def errors(self) -> FetchingIterator:
+        if self.status() != TaskStatus.SUBMITTED and self.status() != TaskStatus.RUNNING:
+            if self.errors_iterator:
+                return self.errors_iterator
+
+            self.errors_iterator = FetchingIterator()
+        return None
+ 
+    @property
+    def kind(self) -> TaskType:
+        return self.task_type