Improving composite annotations uploads process

Author: mmerce

HISTORY.rst

@@ -3,6 +3,12 @@
 History
 -------
 
+9.8.2 (2025-03-21)
+------------------
+
+- Retrying annotations update to avoid temporary concurrency issues in
+  source composites updates.
+
 9.8.1 (2025-01-14)
 ------------------
 

bigml/api_handlers/sourcehandler.py

@@ -24,6 +24,8 @@
 import sys
 import os
 import numbers
+import time
+import logging
 
 from urllib import parse
 
@@ -67,8 +69,13 @@
 from bigml.api_handlers.resourcehandler import ResourceHandlerMixin, LOGGER
 from bigml.fields import Fields
 
+LOG_FORMAT = '%(asctime)-15s: %(message)s'
+LOGGER = logging.getLogger('BigML')
+CONSOLE = logging.StreamHandler()
+CONSOLE.setLevel(logging.WARNING)
+LOGGER.addHandler(CONSOLE)
 
-MAX_CHANGES = 500
+MAX_CHANGES = 5
 
 
 def compact_regions(regions):
@@ -508,6 +515,8 @@ def update_composite_annotations(self, source, images_file,
             try:
                 _ = file_list.index(filename)
             except ValueError:
+                LOGGER.error("WARNING: Could not find annotated file (%s)"
+                             " in the composite's sources list", filename)
                 continue
             for key in annotation.keys():
                 if key == "file":
@@ -550,16 +559,33 @@ def update_composite_annotations(self, source, images_file,
                              "value": value,
                              "components": [source_id]})
             except Exception:
+                LOGGER.error("WARNING: Problem adding annotation to %s (%s)",
+                             field, values)
                 pass
 
         # we need to limit the amount of changes per update
         batches_number = int(len(changes) / MAX_CHANGES)
         for offset in range(0, batches_number + 1):
-            new_batch = changes[offset * MAX_CHANGES: (offset + 1) * MAX_CHANGES]
+            new_batch = changes[
+                offset * MAX_CHANGES: (offset + 1) * MAX_CHANGES]
             if new_batch:
                 source = self.update_source(source,
                                             {"row_values": new_batch})
-                self.ok(source)
+                if source["error"] is not None:
+                    # retrying in case update is temporarily unavailable
+                    time.sleep(1)
+                    source = self.get_source(source)
+                    self.ok(source)
+                    source = self.update_source(source,
+                                                {"row_values": new_batch})
+                    if source["error"] is not None:
+                        LOGGER.error("WARNING: Some annotations were not"
+                                     " updated (%s)",
+                                      new_batch)
+                if not self.ok(source):
+                    raise Exception(
+                        f"Failed to update {len(new_batch)} annotations.")
+                time.sleep(0.1)
 
         return source
 

bigml/bigmlconnection.py

@@ -406,7 +406,7 @@ def _create(self, url, body, verify=None, organization=None):
                     error = json_load(response.content)
                     LOGGER.error(self.error_message(error, method='create'))
                 elif code != HTTP_ACCEPTED:
-                    LOGGER.error("Unexpected error (%s)", code)
+                    LOGGER.error("CREATE Unexpected error (%s)", code)
                     code = HTTP_INTERNAL_SERVER_ERROR
             except ValueError as exc:
                 LOGGER.error("Malformed response: %s", str(exc))
@@ -489,7 +489,7 @@ def _get(self, url, query_string='',
                 LOGGER.error(self.error_message(error, method='get',
                              resource_id=resource_id))
             else:
-                LOGGER.error("Unexpected error (%s)", code)
+                LOGGER.error("GET Unexpected error (%s)", code)
                 code = HTTP_INTERNAL_SERVER_ERROR
 
         except ValueError as exc:
@@ -582,7 +582,7 @@ def _list(self, url, query_string='', organization=None):
                           HTTP_TOO_MANY_REQUESTS]:
                 error = json_load(response.content)
             else:
-                LOGGER.error("Unexpected error (%s)", code)
+                LOGGER.error("LIST Unexpected error (%s)", code)
                 code = HTTP_INTERNAL_SERVER_ERROR
         except ValueError as exc:
             LOGGER.error("Malformed response: %s", str(exc))
@@ -662,7 +662,7 @@ def _update(self, url, body, organization=None, resource_id=None):
                 LOGGER.error(self.error_message(error, method='update',
                              resource_id=resource_id))
             else:
-                LOGGER.error("Unexpected error (%s)", code)
+                LOGGER.error("UPDATE Unexpected error (%s)", code)
                 code = HTTP_INTERNAL_SERVER_ERROR
         except ValueError:
             LOGGER.error("Malformed response")

bigml/version.py

@@ -1 +1 @@
-__version__ = '9.8.1'
+__version__ = '9.8.2'