From 4aac20cabe582e7081a9f9375bf3618deee5f573 Mon Sep 17 00:00:00 2001 From: claw Date: Thu, 21 May 2026 07:33:29 +0000 Subject: [PATCH 01/12] feat: enhance error logging and add error_handler to queries --- aperturedb/CommonLibrary.py | 17 ++++++++++++++--- aperturedb/ParallelQuery.py | 9 ++++++++- aperturedb/ParallelQuerySet.py | 6 ++++-- 3 files changed, 26 insertions(+), 6 deletions(-) diff --git a/aperturedb/CommonLibrary.py b/aperturedb/CommonLibrary.py index 24791199..8b2e2452 100644 --- a/aperturedb/CommonLibrary.py +++ b/aperturedb/CommonLibrary.py @@ -270,7 +270,8 @@ def execute_query(client: Connector, query: Commands, blobs: Blobs = [], success_statuses: list[int] = [0], response_handler: Optional[Callable] = None, commands_per_query: int = 1, blobs_per_query: int = 0, - strict_response_validation: bool = False, cmd_index=None) -> Tuple[int, CommandResponses, Blobs]: + strict_response_validation: bool = False, cmd_index=None, + error_handler: Optional[Callable] = None) -> Tuple[int, CommandResponses, Blobs]: """ Execute a batch of queries, doing useful logging around it. Calls the response handler if provided. @@ -313,7 +314,7 @@ def execute_query(client: Connector, query: Commands, raise e else: # Transaction failed entirely. - logger.error(f"Failed query = {query} with response = {r}") + logger.error(f"Transaction failed. Response: {r}") result = 1 statuses = {} @@ -337,9 +338,19 @@ def execute_query(client: Connector, query: Commands, warn_list.append(wr) if len(warn_list) != 0: logger.warning( - f"Partial errors:\r\n{json.dumps(query, default=str)}\r\n{json.dumps(warn_list, default=str)}") + f"Encountered {len(warn_list)} partial errors. " + "Use error_handler or inspect response for details." + ) result = 2 + if result != 0 and error_handler is not None: + try: + error_handler(query, r, b) + except BaseException as e: + logger.exception(e) + if strict_response_validation: + raise e + return result, r, b diff --git a/aperturedb/ParallelQuery.py b/aperturedb/ParallelQuery.py index 9853b878..d015b9c1 100644 --- a/aperturedb/ParallelQuery.py +++ b/aperturedb/ParallelQuery.py @@ -158,9 +158,12 @@ def process_responses(requests, input_blobs, responses, output_blobs): worker_stats = {} if not self.dry_run: response_handler = None + error_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, "error_handler") and callable(self.generator.error_handler): + error_handler = self.generator.error_handler if hasattr(self.generator, "strict_response_validation") and isinstance(self.generator.strict_response_validation, bool): strict_response_validation = self.generator.strict_response_validation @@ -185,7 +188,8 @@ def response_handler(query, qblobs, resp, rblobs, qindex): return indexless_hand self.commands_per_query, self.blobs_per_query, strict_response_validation=strict_response_validation, - cmd_index=batch_start) + cmd_index=batch_start, + error_handler=error_handler) if result == 0: query_time = client.get_last_query_time() worker_stats["succeeded_commands"] = len(q) @@ -252,6 +256,9 @@ def worker(self, thid: int, generator, start: int, end: int, run_event) -> None: if self.stats: self.pb.update(batch_end - batch_start) logger.info(f"Worker {thid} executed {total_batches} batches") + # Explicitly delete the client to ensure the connection is closed immediately + if client is not None: + del client def get_objects_existed(self) -> int: return sum([stat["objects_existed"] diff --git a/aperturedb/ParallelQuerySet.py b/aperturedb/ParallelQuerySet.py index 119379b7..87958664 100644 --- a/aperturedb/ParallelQuerySet.py +++ b/aperturedb/ParallelQuerySet.py @@ -49,7 +49,8 @@ def gen_execute_batch_sets(base_executor): def execute_batch_sets(client, query_set, blob_set, success_statuses: list[int] = [0], response_handler: Optional[Callable] = None, commands_per_query: list[int] = -1, blobs_per_query: list[int] = -1, - strict_response_validation: bool = False, cmd_index: int = None): + strict_response_validation: bool = False, cmd_index: int = None, + error_handler: Optional[Callable] = None): logger.info("Execute Batch Sets = Batch Size {0} Comands Per Query {1} Blobs Per Query {2}".format( len(query_set), commands_per_query, blobs_per_query)) @@ -303,7 +304,8 @@ def constraint_filter(single_line, single_results): commands_per_query[i], blobs_per_query[i], strict_response_validation=strict_response_validation, - cmd_index=cmd_index) + cmd_index=cmd_index, + error_handler=error_handler) if response_handler != None and client.last_query_ok(): def map_to_set(query, query_blobs, resp, resp_blobs): response_handler( From 52a5704bd81890008f20f394e94a346718d4c580 Mon Sep 17 00:00:00 2001 From: ad-claw000 Date: Thu, 21 May 2026 08:01:48 +0000 Subject: [PATCH 02/12] Address Copilot review comments on PR #727 --- aperturedb/CommonLibrary.py | 2 ++ aperturedb/ParallelQuery.py | 3 --- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/aperturedb/CommonLibrary.py b/aperturedb/CommonLibrary.py index 8b2e2452..47b34201 100644 --- a/aperturedb/CommonLibrary.py +++ b/aperturedb/CommonLibrary.py @@ -288,6 +288,8 @@ def execute_query(client: Connector, query: Commands, commands_per_query (int, optional): The number of commands per query. Defaults to 1. blobs_per_query (int, optional): The number of blobs per query. Defaults to 0. strict_response_validation (bool, optional): Whether to strictly validate the response. Defaults to False. + cmd_index (int, optional): The index of the command or batch, passed to the response handler. Defaults to None. + error_handler (Callable, optional): Callback invoked when the query returns an unexpected status or fails. Expected signature is `error_handler(query, response, blobs)`. Defaults to None. Returns: int: The result code. diff --git a/aperturedb/ParallelQuery.py b/aperturedb/ParallelQuery.py index d015b9c1..8b4bf15a 100644 --- a/aperturedb/ParallelQuery.py +++ b/aperturedb/ParallelQuery.py @@ -256,9 +256,6 @@ def worker(self, thid: int, generator, start: int, end: int, run_event) -> None: if self.stats: self.pb.update(batch_end - batch_start) logger.info(f"Worker {thid} executed {total_batches} batches") - # Explicitly delete the client to ensure the connection is closed immediately - if client is not None: - del client def get_objects_existed(self) -> int: return sum([stat["objects_existed"] From 44dc1aa2a806346fe97c45485f6abc85a3c5dc1b Mon Sep 17 00:00:00 2001 From: ad-claw000 Date: Thu, 21 May 2026 15:40:04 +0000 Subject: [PATCH 03/12] Address review comments for error_handler exception handling and tests - Switched from BaseException to Exception in response/error handlers - Re-raised using bare raise to preserve traceback context - Added monkeypatch tests for error_handler scenarios --- aperturedb/CommonLibrary.py | 8 +-- test/test_ResponseHandler.py | 98 ++++++++++++++++++++++++++++++++++++ 2 files changed, 102 insertions(+), 4 deletions(-) diff --git a/aperturedb/CommonLibrary.py b/aperturedb/CommonLibrary.py index 2d69a074..eeae70a5 100644 --- a/aperturedb/CommonLibrary.py +++ b/aperturedb/CommonLibrary.py @@ -313,10 +313,10 @@ def execute_query(client: Connector, query: Commands, map_response_to_handler(response_handler, query, blobs, r, b, commands_per_query, blobs_per_query, cmd_index) - except BaseException as e: + except Exception as e: logger.exception(e) if strict_response_validation: - raise e + raise else: # Transaction failed entirely. logger.error( @@ -352,10 +352,10 @@ def execute_query(client: Connector, query: Commands, if result != 0 and error_handler is not None: try: error_handler(query, r, b) - except BaseException as e: + except Exception as e: logger.exception(e) if strict_response_validation: - raise e + raise return result, r, b diff --git a/test/test_ResponseHandler.py b/test/test_ResponseHandler.py index 91b6bb9d..6741ea99 100644 --- a/test/test_ResponseHandler.py +++ b/test/test_ResponseHandler.py @@ -488,3 +488,101 @@ def mock_query(self, request, blobs): querier.query(generator, numthreads=1, batchsize=2, stats=False) assert len(changed_ids) == 2 + def test_error_handler_total_failure(self, db, monkeypatch): + from aperturedb.QueryGenerator import QueryGenerator + from aperturedb.ParallelQuery import ParallelQuery + from aperturedb.Connector import Connector + + class QGErrorHandlerFailure(QueryGenerator): + def __init__(self): + self.error_handler_called = False + + def __len__(self): + return 1 + + def getitem(self, idx): + return [{"FindImage": {}}], [] + + def error_handler(self, q, r, b): + self.error_handler_called = True + + def mock_query(self, request, blobs): + self.response = [{"status": -1}] + self.blobs = [] + return self.response, [] + + monkeypatch.setattr(Connector, "query", mock_query) + monkeypatch.setattr(Connector, "last_query_ok", lambda self: False) + monkeypatch.setattr(Connector, "clone", lambda self: self) + + generator = QGErrorHandlerFailure() + querier = ParallelQuery(db) + querier.query(generator, numthreads=1, batchsize=1, stats=False) + + assert generator.error_handler_called + + def test_error_handler_partial_error(self, db, monkeypatch): + from aperturedb.QueryGenerator import QueryGenerator + from aperturedb.ParallelQuery import ParallelQuery + from aperturedb.Connector import Connector + + class QGErrorHandlerPartial(QueryGenerator): + def __init__(self): + self.error_handler_called = False + + def __len__(self): + return 1 + + def getitem(self, idx): + return [{"FindImage": {}}, {"AddImage": {}}], [] + + def error_handler(self, q, r, b): + self.error_handler_called = True + + def mock_query(self, request, blobs): + self.response = [{"FindImage": {"status": 0}}, {"AddImage": {"status": 2}}] + self.blobs = [] + return self.response, [] + + monkeypatch.setattr(Connector, "query", mock_query) + monkeypatch.setattr(Connector, "last_query_ok", lambda self: True) + monkeypatch.setattr(Connector, "clone", lambda self: self) + + generator = QGErrorHandlerPartial() + querier = ParallelQuery(db) + querier.query(generator, numthreads=1, batchsize=1, stats=False) + + assert generator.error_handler_called + + def test_error_handler_success(self, db, monkeypatch): + from aperturedb.QueryGenerator import QueryGenerator + from aperturedb.ParallelQuery import ParallelQuery + from aperturedb.Connector import Connector + + class QGErrorHandlerSuccess(QueryGenerator): + def __init__(self): + self.error_handler_called = False + + def __len__(self): + return 1 + + def getitem(self, idx): + return [{"FindImage": {}}, {"AddImage": {}}], [] + + def error_handler(self, q, r, b): + self.error_handler_called = True + + def mock_query(self, request, blobs): + self.response = [{"FindImage": {"status": 0}}, {"AddImage": {"status": 0}}] + self.blobs = [] + return self.response, [] + + monkeypatch.setattr(Connector, "query", mock_query) + monkeypatch.setattr(Connector, "last_query_ok", lambda self: True) + monkeypatch.setattr(Connector, "clone", lambda self: self) + + generator = QGErrorHandlerSuccess() + querier = ParallelQuery(db) + querier.query(generator, numthreads=1, batchsize=1, stats=False) + + assert not generator.error_handler_called From 4ce015a06b272f86236c45fa91f8283d4fa88336 Mon Sep 17 00:00:00 2001 From: claw Date: Thu, 21 May 2026 16:18:23 +0000 Subject: [PATCH 04/12] style: apply pre-commit autopep8 fixes --- test/test_ResponseHandler.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/test/test_ResponseHandler.py b/test/test_ResponseHandler.py index 6741ea99..eeb273e3 100644 --- a/test/test_ResponseHandler.py +++ b/test/test_ResponseHandler.py @@ -488,6 +488,7 @@ def mock_query(self, request, blobs): querier.query(generator, numthreads=1, batchsize=2, stats=False) assert len(changed_ids) == 2 + def test_error_handler_total_failure(self, db, monkeypatch): from aperturedb.QueryGenerator import QueryGenerator from aperturedb.ParallelQuery import ParallelQuery @@ -540,7 +541,8 @@ def error_handler(self, q, r, b): self.error_handler_called = True def mock_query(self, request, blobs): - self.response = [{"FindImage": {"status": 0}}, {"AddImage": {"status": 2}}] + self.response = [{"FindImage": {"status": 0}}, + {"AddImage": {"status": 2}}] self.blobs = [] return self.response, [] @@ -573,7 +575,8 @@ def error_handler(self, q, r, b): self.error_handler_called = True def mock_query(self, request, blobs): - self.response = [{"FindImage": {"status": 0}}, {"AddImage": {"status": 0}}] + self.response = [{"FindImage": {"status": 0}}, + {"AddImage": {"status": 0}}] self.blobs = [] return self.response, [] From fef7e0c148330cca307f2a550b1a1bb0aa934b7f Mon Sep 17 00:00:00 2001 From: claw Date: Thu, 21 May 2026 16:28:53 +0000 Subject: [PATCH 05/12] fix: restore working code blocks and fix UnboundLocalErrors - restore 'with open' in Sources.py and VideoDataCSV.py to fix UnboundLocalError - restore 'with iolock' in cli/mount_coco.py to avoid deadlock on exception - revert broken DaskManager.py changes which removed 'raise' causing UnboundLocalError - restore dry_run kwarg pass in ParallelQuery.py for DaskManager --- aperturedb/DaskManager.py | 18 +++++++++++++----- aperturedb/ParallelQuery.py | 2 +- aperturedb/Sources.py | 8 ++------ aperturedb/VideoDataCSV.py | 5 ++--- aperturedb/cli/mount_coco.py | 10 +++++++--- 5 files changed, 25 insertions(+), 18 deletions(-) diff --git a/aperturedb/DaskManager.py b/aperturedb/DaskManager.py index 96fa1f87..6a79d0c5 100644 --- a/aperturedb/DaskManager.py +++ b/aperturedb/DaskManager.py @@ -48,8 +48,8 @@ def __del__(self): self._client.close() self._cluster.close() - def run(self, QueryClass: type[ParallelQuery], client: Connector, generator, batchsize, stats): - def process(df, host, port, use_ssl, ca_cert, verify_hostname, session, connnector_type): + def run(self, QueryClass: type[ParallelQuery], client: Connector, generator, batchsize, stats, dry_run=False): + def process(df, host, port, use_ssl, ca_cert, verify_hostname, session, connector_type, dry_run): metrics = Stats() # Dask reads data in partitions, and the first partition is of 2 rows, with all # values as 'foo'. This is for sampling the column names and types. Should not process @@ -62,7 +62,7 @@ def process(df, host, port, use_ssl, ca_cert, verify_hostname, session, connnect shared_data = SimpleNamespace() shared_data.session = session shared_data.lock = Lock() - client = connnector_type( + client = connector_type( host=host, port=port, use_ssl=use_ssl, ca_cert=ca_cert, @@ -70,8 +70,15 @@ def process(df, host, port, use_ssl, ca_cert, verify_hostname, session, connnect shared_data=shared_data) except Exception as e: logger.exception(e) + raise #from aperturedb.ParallelLoader import ParallelLoader - loader = QueryClass(client) + import inspect + sig = inspect.signature(QueryClass.__init__) + if "dry_run" in sig.parameters or any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()): + loader = QueryClass(client, dry_run=dry_run) + else: + loader = QueryClass(client) + loader.dry_run = dry_run for i in range(0, len(df), batchsize): end = min(i + batchsize, len(df)) slice = df[i:end] @@ -101,7 +108,8 @@ def process(df, host, port, use_ssl, ca_cert, verify_hostname, session, connnect client.config.ca_cert, client.config.verify_hostname, client.shared_data.session, - type(client)) + type(client), + dry_run) computation = computation.persist() if stats: progress(computation) diff --git a/aperturedb/ParallelQuery.py b/aperturedb/ParallelQuery.py index 8b4bf15a..4931a9c2 100644 --- a/aperturedb/ParallelQuery.py +++ b/aperturedb/ParallelQuery.py @@ -290,7 +290,7 @@ def query(self, generator, batchsize: int = 1, numthreads: int = 4, stats: bool if use_dask: results, self.total_actions_time = self.daskmanager.run( - self.__class__, self.client, generator, batchsize, stats=stats) + self.__class__, self.client, generator, batchsize, stats=stats, dry_run=self.dry_run) self.actual_stats = [] for result in results: if result is not None: diff --git a/aperturedb/Sources.py b/aperturedb/Sources.py index a652d28f..34a2650c 100644 --- a/aperturedb/Sources.py +++ b/aperturedb/Sources.py @@ -30,16 +30,12 @@ def load_from_file(self, filename): Load data from a file. """ try: - fd = open(filename, "rb") - buff = fd.read() - fd.close() + with open(filename, "rb") as fd: + buff = fd.read() return True, buff except Exception as e: logger.error(f"VALIDATION ERROR: {filename}") logger.exception(e) - finally: - if not fd.closed: - fd.close() return False, None def load_from_http_url(self, url, validator): diff --git a/aperturedb/VideoDataCSV.py b/aperturedb/VideoDataCSV.py index 99868a30..750b8797 100644 --- a/aperturedb/VideoDataCSV.py +++ b/aperturedb/VideoDataCSV.py @@ -137,9 +137,8 @@ def load_video(self, filename): logger.exception(e) try: - fd = open(filename, "rb") - buff = fd.read() - fd.close() + with open(filename, "rb") as fd: + buff = fd.read() return True, buff except Exception as e: logger.error(f"Video Error: {filename}") diff --git a/aperturedb/cli/mount_coco.py b/aperturedb/cli/mount_coco.py index a79936a0..c3761101 100644 --- a/aperturedb/cli/mount_coco.py +++ b/aperturedb/cli/mount_coco.py @@ -232,25 +232,29 @@ def read(self, path, size, offset): logger.info(f"path = {path}, size = {size}, offset = {offset}") filename = os.path.basename(path) logger.info(f"Filename = {filename}") + img = b'' + slen = 0 + if filename == meta_info: try: img = self.meta_data slen = len(img) except Exception as e: logger.exception(e) + return -errno.EIO else: try: image_id = filename[:-4] idx = self._images.images_ids.index(image_id) logger.info(f"Image id = {image_id}, index = {idx}") - self.iolock.acquire() - img = self._images.get_image_by_index(idx) - self.iolock.release() + with self.iolock: + img = self._images.get_image_by_index(idx) slen = len(img) logger.debug(f"Type = {type(img)}, len = {slen}") except Exception as e: logger.exception(e) logger.error("Error occured") + return -errno.EIO if offset < slen: if offset + size > slen: From 8578978067085a4ce1aff81dae6853c803f902a3 Mon Sep 17 00:00:00 2001 From: ad-claw000 Date: Thu, 21 May 2026 16:33:42 +0000 Subject: [PATCH 06/12] test: correct mocked response formats for error_handler tests --- test/test_ResponseHandler.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/test_ResponseHandler.py b/test/test_ResponseHandler.py index eeb273e3..3d01368a 100644 --- a/test/test_ResponseHandler.py +++ b/test/test_ResponseHandler.py @@ -508,7 +508,7 @@ def error_handler(self, q, r, b): self.error_handler_called = True def mock_query(self, request, blobs): - self.response = [{"status": -1}] + self.response = {"status": -1, "info": "Transaction failed"} self.blobs = [] return self.response, [] @@ -542,7 +542,7 @@ def error_handler(self, q, r, b): def mock_query(self, request, blobs): self.response = [{"FindImage": {"status": 0}}, - {"AddImage": {"status": 2}}] + {"AddImage": {"status": 1}}] self.blobs = [] return self.response, [] From 3467faf8c92dfa42b3ab63732420551c6d2b7004 Mon Sep 17 00:00:00 2001 From: claw Date: Sat, 23 May 2026 06:40:27 +0000 Subject: [PATCH 07/12] fix: address review comments on warn_list and typing --- aperturedb/CommonLibrary.py | 9 ++++----- aperturedb/ParallelQuerySet.py | 4 ++-- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/aperturedb/CommonLibrary.py b/aperturedb/CommonLibrary.py index eeae70a5..82422068 100644 --- a/aperturedb/CommonLibrary.py +++ b/aperturedb/CommonLibrary.py @@ -337,14 +337,13 @@ def execute_query(client: Connector, query: Commands, # last_query_ok means result status >= 0 if result != 1: - warn_list = [] + warnings_count = 0 for status, results in statuses.items(): if status not in success_statuses: - for wr in results: - warn_list.append(wr) - if len(warn_list) != 0: + warnings_count += len(results) + if warnings_count != 0: logger.warning( - f"Encountered {len(warn_list)} partial errors. " + f"Encountered {warnings_count} partial errors. " "Use error_handler or inspect response for details." ) result = 2 diff --git a/aperturedb/ParallelQuerySet.py b/aperturedb/ParallelQuerySet.py index 87958664..acecf574 100644 --- a/aperturedb/ParallelQuerySet.py +++ b/aperturedb/ParallelQuerySet.py @@ -47,8 +47,8 @@ def gen_execute_batch_sets(base_executor): # execution # def execute_batch_sets(client, query_set, blob_set, success_statuses: list[int] = [0], - response_handler: Optional[Callable] = None, commands_per_query: list[int] = -1, - blobs_per_query: list[int] = -1, + response_handler: Optional[Callable] = None, commands_per_query: Optional[list[int]] = None, + blobs_per_query: Optional[list[int]] = None, strict_response_validation: bool = False, cmd_index: int = None, error_handler: Optional[Callable] = None): From f201a4732828b6aa291312d44e03b6a4a65a481e Mon Sep 17 00:00:00 2001 From: claw Date: Sat, 23 May 2026 11:50:33 +0000 Subject: [PATCH 08/12] fix: initialize default list for commands_per_query and blobs_per_query --- aperturedb/ParallelQuerySet.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/aperturedb/ParallelQuerySet.py b/aperturedb/ParallelQuerySet.py index acecf574..e94b34ae 100644 --- a/aperturedb/ParallelQuerySet.py +++ b/aperturedb/ParallelQuerySet.py @@ -64,6 +64,11 @@ def execute_batch_sets(client, query_set, blob_set, success_statuses: list[int] raise Exception("Query set must be a list of lists") set_total = len(first_element) + if commands_per_query is None: + commands_per_query = [1] * set_total + if blobs_per_query is None: + blobs_per_query = [0] * set_total + # Check if blobs are a simple array or nested array of blobs per_set_blobs = isinstance(blob_set, list) and len( blob_set) > 0 and isinstance(blob_set[0], list) From 1a19100e08905d59748bd67a728178314462a8ad Mon Sep 17 00:00:00 2001 From: claw Date: Sat, 23 May 2026 17:56:07 +0000 Subject: [PATCH 09/12] fix: address remaining review comments for error_handler and success_statuses --- aperturedb/CommonLibrary.py | 2 +- aperturedb/ParallelQuerySet.py | 7 +++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/aperturedb/CommonLibrary.py b/aperturedb/CommonLibrary.py index 82422068..6621d084 100644 --- a/aperturedb/CommonLibrary.py +++ b/aperturedb/CommonLibrary.py @@ -350,7 +350,7 @@ def execute_query(client: Connector, query: Commands, if result != 0 and error_handler is not None: try: - error_handler(query, r, b) + error_handler(query, r, blobs) except Exception as e: logger.exception(e) if strict_response_validation: diff --git a/aperturedb/ParallelQuerySet.py b/aperturedb/ParallelQuerySet.py index e94b34ae..c82bd140 100644 --- a/aperturedb/ParallelQuerySet.py +++ b/aperturedb/ParallelQuerySet.py @@ -46,12 +46,15 @@ def gen_execute_batch_sets(base_executor): # if blob_set is a list of lists, each inner list will be given to the inner # execution # - def execute_batch_sets(client, query_set, blob_set, success_statuses: list[int] = [0], + def execute_batch_sets(client, query_set, blob_set, success_statuses: Optional[list[int]] = None, response_handler: Optional[Callable] = None, commands_per_query: Optional[list[int]] = None, blobs_per_query: Optional[list[int]] = None, strict_response_validation: bool = False, cmd_index: int = None, error_handler: Optional[Callable] = None): + if success_statuses is None: + success_statuses = [0] + logger.info("Execute Batch Sets = Batch Size {0} Comands Per Query {1} Blobs Per Query {2}".format( len(query_set), commands_per_query, blobs_per_query)) @@ -279,7 +282,7 @@ def constraint_filter(single_line, single_results): query_filter = constraint_filter - local_success_statuses = [0, 2] + local_success_statuses = success_statuses # queries are by row first, so we run query_filter on each query # we pass the entire row's data, then we retrieve all of the stored results for that row From 373d387ca30999803bd907f5f672782b6612fddb Mon Sep 17 00:00:00 2001 From: claw Date: Sun, 24 May 2026 02:33:38 +0000 Subject: [PATCH 10/12] fix: default success_statuses to ParallelQuery.success_statuses --- aperturedb/ParallelQuerySet.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aperturedb/ParallelQuerySet.py b/aperturedb/ParallelQuerySet.py index c82bd140..4eaaecde 100644 --- a/aperturedb/ParallelQuerySet.py +++ b/aperturedb/ParallelQuerySet.py @@ -53,7 +53,7 @@ def execute_batch_sets(client, query_set, blob_set, success_statuses: Optional[l error_handler: Optional[Callable] = None): if success_statuses is None: - success_statuses = [0] + success_statuses = ParallelQuery.success_statuses logger.info("Execute Batch Sets = Batch Size {0} Comands Per Query {1} Blobs Per Query {2}".format( len(query_set), commands_per_query, blobs_per_query)) From 45d0f95932f48d634b2502614b744f128d67dc99 Mon Sep 17 00:00:00 2001 From: claw Date: Sun, 24 May 2026 11:15:04 +0000 Subject: [PATCH 11/12] fix: address low confidence review comments regarding logging - Move logging of commands_per_query and blobs_per_query down so they don't log None - Truncate the failed query in the total transaction failure error log --- aperturedb/CommonLibrary.py | 7 ++++++- aperturedb/ParallelQuerySet.py | 8 ++++---- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/aperturedb/CommonLibrary.py b/aperturedb/CommonLibrary.py index fcb0994b..349a757c 100644 --- a/aperturedb/CommonLibrary.py +++ b/aperturedb/CommonLibrary.py @@ -319,8 +319,13 @@ def execute_query(client: Connector, query: Commands, raise else: # Transaction failed entirely. + num_commands = len(query) if isinstance(query, list) else 1 + truncated_query = query[:2] if isinstance( + query, list) and num_commands > 2 else query + query_summary = f"{num_commands} commands (showing first 2: { + truncated_query})" if num_commands > 2 else str(query) logger.error( - f"Failed query = {query} with response = {censor_tokens(r)}") + f"Failed query = {query_summary} with response = {censor_tokens(r)}") result = 1 statuses = {} diff --git a/aperturedb/ParallelQuerySet.py b/aperturedb/ParallelQuerySet.py index 4eaaecde..9f0f3cd5 100644 --- a/aperturedb/ParallelQuerySet.py +++ b/aperturedb/ParallelQuerySet.py @@ -55,15 +55,12 @@ def execute_batch_sets(client, query_set, blob_set, success_statuses: Optional[l if success_statuses is None: success_statuses = ParallelQuery.success_statuses - logger.info("Execute Batch Sets = Batch Size {0} Comands Per Query {1} Blobs Per Query {2}".format( - len(query_set), commands_per_query, blobs_per_query)) - batch_size = len(query_set) # test query set first_element = query_set[0] if not isinstance(first_element, list): - logger.error("First Element not a list: {first_element}") + logger.error(f"First Element not a list: {first_element}") raise Exception("Query set must be a list of lists") set_total = len(first_element) @@ -72,6 +69,9 @@ def execute_batch_sets(client, query_set, blob_set, success_statuses: Optional[l if blobs_per_query is None: blobs_per_query = [0] * set_total + logger.info("Execute Batch Sets = Batch Size {0} Comands Per Query {1} Blobs Per Query {2}".format( + batch_size, commands_per_query, blobs_per_query)) + # Check if blobs are a simple array or nested array of blobs per_set_blobs = isinstance(blob_set, list) and len( blob_set) > 0 and isinstance(blob_set[0], list) From 7b66be26fc795c716f17a458de99a007a8fb8568 Mon Sep 17 00:00:00 2001 From: claw Date: Sun, 24 May 2026 14:09:23 +0000 Subject: [PATCH 12/12] fix: address review comments regarding f-string formatting, typo, and typing --- aperturedb/CommonLibrary.py | 6 ++++-- aperturedb/ParallelQuerySet.py | 4 ++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/aperturedb/CommonLibrary.py b/aperturedb/CommonLibrary.py index 349a757c..ab91794a 100644 --- a/aperturedb/CommonLibrary.py +++ b/aperturedb/CommonLibrary.py @@ -322,8 +322,10 @@ def execute_query(client: Connector, query: Commands, num_commands = len(query) if isinstance(query, list) else 1 truncated_query = query[:2] if isinstance( query, list) and num_commands > 2 else query - query_summary = f"{num_commands} commands (showing first 2: { - truncated_query})" if num_commands > 2 else str(query) + query_summary = ( + f"{num_commands} commands (showing first 2: {truncated_query})" + if num_commands > 2 else str(query) + ) logger.error( f"Failed query = {query_summary} with response = {censor_tokens(r)}") result = 1 diff --git a/aperturedb/ParallelQuerySet.py b/aperturedb/ParallelQuerySet.py index 9f0f3cd5..9a6e885c 100644 --- a/aperturedb/ParallelQuerySet.py +++ b/aperturedb/ParallelQuerySet.py @@ -49,7 +49,7 @@ def gen_execute_batch_sets(base_executor): def execute_batch_sets(client, query_set, blob_set, success_statuses: Optional[list[int]] = None, response_handler: Optional[Callable] = None, commands_per_query: Optional[list[int]] = None, blobs_per_query: Optional[list[int]] = None, - strict_response_validation: bool = False, cmd_index: int = None, + strict_response_validation: bool = False, cmd_index: Optional[int] = None, error_handler: Optional[Callable] = None): if success_statuses is None: @@ -69,7 +69,7 @@ def execute_batch_sets(client, query_set, blob_set, success_statuses: Optional[l if blobs_per_query is None: blobs_per_query = [0] * set_total - logger.info("Execute Batch Sets = Batch Size {0} Comands Per Query {1} Blobs Per Query {2}".format( + logger.info("Execute Batch Sets = Batch Size {0} Commands Per Query {1} Blobs Per Query {2}".format( batch_size, commands_per_query, blobs_per_query)) # Check if blobs are a simple array or nested array of blobs