Merge pull request #247 from aperture-data/release-0.4.6

Release 0.4.6

Author: gsaluja9

aperturedb/CSVParser.py

@@ -31,11 +31,20 @@ class CSVParser(Subscriptable):
         In Dask mode the CSV file is read in chunks, and the operations are performed on each chunk.
         The tricky bit is that the chunck size is not known till the loader is created, so the processing happens when ingest is called.
         So the Data CSV has another signature, where the df is passed explicitly.
+
+    Typically, the response_handler is application specific, and loading does not break
+    on errors in response_handlers, so the default behaviour is to log the error and continue.
+    If you want to break on errors, set strict_response_validation to True.
     """
 
-    def __init__(self, filename, df=None, use_dask=False):
+    def __init__(self,
+                 filename: str,
+                 df=None,
+                 use_dask: bool = False,
+                 strict_response_validation: bool = False):
         self.use_dask = use_dask
         self.filename = filename
+        self.strict_response_validation = strict_response_validation
 
         if not use_dask:
             if df is None:

aperturedb/ConnectionDataCSV.py

@@ -1,6 +1,6 @@
 import logging
 from aperturedb.CSVParser import CSVParser, CONSTRAINTS_PREFIX
-from aperturedb.Query import QueryBuilder
+from aperturedb.Query import QueryBuilder, ObjectType
 
 logger = logging.getLogger(__name__)
 
@@ -83,6 +83,7 @@ def getitem(self, idx):
         dst_value = self.df.loc[idx, self.header[2]]
         connection_class = self.df.loc[idx, CONNECTION_CLASS]
         q = []
+        members = ["_Image", "_Blob", "_Video", "_Descriptor"]
 
         try:
             ref_src = (2 * idx) % 100000 + 1
@@ -93,6 +94,9 @@ def getitem(self, idx):
                     self.src_key: ["==", src_value]
                 }
             }
+            # Special case for objects with blobs
+            if self.src_class in members:
+                cmd_params["blobs"] = False
             q.append(QueryBuilder.find_command(self.src_class, cmd_params))
 
             ref_dst = ref_src + 1
@@ -103,6 +107,9 @@ def getitem(self, idx):
                     self.dst_key: ["==", dst_value]
                 }
             }
+            # Special case for objects with blobs
+            if self.dst_class in members:
+                cmd_params["blobs"] = False
             q.append(QueryBuilder.find_command(self.dst_class, cmd_params))
 
             ae = self._basic_command(idx,

aperturedb/Images.py

@@ -418,7 +418,6 @@ def get_similar_images(self, set_name, n_neighbors):
                     "k_neighbors": n_neighbors + 1,
                     "blobs":     False,
                     "distances": True,
-                    "ids":       True,
                     "uniqueids": True,
                 }
             }, {

aperturedb/ParallelQuery.py

@@ -14,7 +14,8 @@
 
 
 def execute_batch(q, blobs, db, success_statuses: list[int] = [0],
-                  response_handler: Callable = None, commands_per_query: int = 1, blobs_per_query: int = 0):
+                  response_handler: Callable = None, commands_per_query: int = 1, blobs_per_query: int = 0,
+                  strict_response_validation: bool = False):
     """
     Execute a batch of queries, doing useful logging around it.
     Calls the response handler if provided.
@@ -64,6 +65,8 @@ def execute_batch(q, blobs, db, success_statuses: list[int] = [0],
                         b[blobs_returned:blobs_returned + b_count] if len(b) < blobs_returned + b_count else None)
                 except BaseException as e:
                     logger.exception(e)
+                    if strict_response_validation:
+                        raise e
                 blobs_returned += b_count
     else:
         # Transaction failed entirely.
@@ -174,10 +177,20 @@ def process_responses(requests, input_blobs, responses, output_blobs):
         worker_stats = {}
         if not self.dry_run:
             response_handler = None
+            strict_response_validation = False
             if hasattr(self.generator, "response_handler") and callable(self.generator.response_handler):
                 response_handler = self.generator.response_handler
+            if hasattr(self.generator, "strict_response_validation") and isinstance(self.generator.strict_response_validation, bool):
+                strict_response_validation = self.generator.strict_response_validation
             result, r, b = execute_batch(
-                q, blobs, db, ParallelQuery.success_statuses, response_handler, self.commands_per_query, self.blobs_per_query)
+                q,
+                blobs,
+                db,
+                ParallelQuery.success_statuses,
+                response_handler,
+                self.commands_per_query,
+                self.blobs_per_query,
+                strict_response_validation=strict_response_validation)
             if result == 0:
                 query_time = db.get_last_query_time()
                 worker_stats["suceeded_commands"] = len(q)

aperturedb/Subscriptable.py

@@ -5,9 +5,10 @@ class Wrapper():
     that will be a part of generator.
     """
 
-    def __init__(self, list, response_handler):
+    def __init__(self, list, response_handler, strict_response_validation):
         self.list = list
         self.response_handler = response_handler
+        self.strict_response_validation = strict_response_validation
 
     def __len__(self):
         return len(self.list)
@@ -29,7 +30,11 @@ def __getitem__(self, subscript):
             step = subscript.step if subscript.step else 1
             return Wrapper(
                 [self.getitem(i) for i in range(start, stop, step)],
-                self.response_handler if hasattr(self, "response_handler") else None)
+                self.response_handler if hasattr(
+                    self, "response_handler") else None,
+                self.strict_response_validation if hasattr(
+                    self, "strict_response_validation") else None
+            )
         else:
             if subscript < len(self):
                 return self.getitem(subscript)

aperturedb/__init__.py

@@ -7,7 +7,7 @@
 
 logger = logging.getLogger(__name__)
 
-__version__ = "0.4.5"
+__version__ = "0.4.6"
 
 # set log level
 logger.setLevel(logging.DEBUG)

setup.py

@@ -22,7 +22,7 @@
 
 setuptools.setup(
     name="aperturedb",
-    version="0.4.5",
+    version="0.4.6",
     description="ApertureDB Client Module",
     install_requires=install_requires,
     long_description=long_description,

test/conftest.py

@@ -51,6 +51,13 @@ def db(request):
     return db
 
 
+def check_response_regressions(queries, input_blobs, responses, output_blobs):
+    # Check that responses have no blobs
+    first_command = list(responses[0].keys())[0]
+    assert "blobs_start" not in responses[0][
+        first_command], f"responses[0]={responses[0]}"
+
+
 @pytest.fixture()
 def insert_data_from_csv(db, request):
     """
@@ -82,6 +89,10 @@ def insert_data_from_csv(in_csv_file, rec_count=-1, expected_error_count=0, load
         if hasattr(request, "param"):
             use_dask = request.param
         data = file_data_pair[in_csv_file](in_csv_file, use_dask=use_dask)
+
+        setattr(data, "response_handler", check_response_regressions)
+        data.strict_response_validation = True
+
         if rec_count != -1:
             data = data[:rec_count]
 
@@ -106,7 +117,7 @@ def insert_data_from_csv(in_csv_file, rec_count=-1, expected_error_count=0, load
         if loader_result_lambda is not None:
             loader_result_lambda(loader, data)
 
-       # Preserve loader so that dask manager is not auto deleted.
+        # Preserve loader so that dask manager is not auto deleted.
         # ---------------
         # Previously, dask's cluster and client were entirely managed in a context
         # insede dask manager. A change upstream broke that, and now we keep the DM

test/docker-compose.yml

@@ -2,7 +2,7 @@ version: '3'
 
 services:
   aperturedb:
-    image: aperturedata/aperturedb
+    image: aperturedata/aperturedb:v0.15.1
     volumes:
       - ./aperturedb:/aperturedb
     ports: