Merge pull request #65 from edanalytics/fix/total_count_logged

Fix: Validation error message shows the wrong number of records

Author: tomreitz

CHANGELOG.md

@@ -1,3 +1,6 @@
+### Unreleased changes
+* bugfix: `validate` and `send` could incorrect number of total records
+
 ### v0.1.6
 <details>
 <summary>Released 2024-11-15</summary>

lightbeam/send.py

@@ -69,7 +69,8 @@ async def do_send(self, endpoint):
         for file_name in data_files:
             with open(file_name) as file:
                 # process each line
-                for line_counter, line in enumerate(file):
+                for i, line in enumerate(file):
+                    line_number = i + 1
                     total_counter += 1
                     data = line.strip()
                     # compute hash of current row
@@ -90,7 +91,7 @@ async def do_send(self, endpoint):
                                         endpoint,
                                         file_name,
                                         data,
-                                        line_counter,
+                                        line_number,
                                         data_hash,
                                     )
                                 )
@@ -104,7 +105,7 @@ async def do_send(self, endpoint):
                         tasks.append(
                             asyncio.create_task(
                                 self.do_post(
-                                    endpoint, file_name, data, line_counter, data_hash
+                                    endpoint, file_name, data, line_number, data_hash
                                 )
                             )
                         )
@@ -136,7 +137,7 @@ async def do_send(self, endpoint):
         })
 
     # Posts a single data payload to a single endpoint
-    async def do_post(self, endpoint, file_name, data, line, data_hash):
+    async def do_post(self, endpoint, file_name, data, line_number, data_hash):
         curr_token_version = int(str(self.lightbeam.token_version))
         while True: # this is not great practice, but an effective way (along with the `break` below) to achieve a do:while loop
             try:
@@ -162,15 +163,15 @@ async def do_post(self, endpoint, file_name, data, line, data_hash):
                             do_append = True
                             for index, item in enumerate(failures):
                                 if item["status_code"]==response.status and item["message"]==message and item["file"]==file_name:
-                                    failures[index]["line_numbers"].append(line)
+                                    failures[index]["line_numbers"].append(line_number)
                                     failures[index]["count"] += 1
                                     do_append = False
                             if do_append:
                                 failure = {
                                     'status_code': response.status,
                                     'message': message,
                                     'file': file_name,
-                                    'line_numbers': [line],
+                                    'line_numbers': [line_number],
                                     'count': 1
                                 }
                                 failures.append(failure)
@@ -208,5 +209,5 @@ async def do_post(self, endpoint, file_name, data, line, data_hash):
             except Exception as e:
                 status = 400
                 self.lightbeam.num_errors += 1
-                self.logger.warn("{0}  (at line {1} of {2} )".format(str(e), line, file_name))
+                self.logger.warn("{0}  (at line {1} of {2} )".format(str(e), line_number, file_name))
                 break

lightbeam/validate.py

@@ -118,12 +118,13 @@ def load_references_data(self, endpoint, references_structure):
         data_files = self.lightbeam.get_data_files_for_endpoint(endpoint)
         for file_name in data_files:
             with open(file_name) as file:
-                for line_counter, line in enumerate(file):
+                for i, line in enumerate(file):
+                    line_number = i + 1
                     line = line.strip()
                     try:
                         payload = json.loads(line)
                     except Exception as e:
-                        self.logger.warning(f"... (ignoring invalid JSON payload at {line_counter} of {file_name})")
+                        self.logger.warning(f"... (ignoring invalid JSON payload at {line_number} of {file_name})")
                     ref_payload = {}
                     for key in references_structure[endpoint]:
                         key = self.EDFI_GENERIC_REFS_TO_PROPERTIES_MAPPING.get(key, {}).get(endpoint, key)
@@ -199,12 +200,13 @@ async def validate_endpoint(self, endpoint):
         for file_name in data_files:
             self.logger.info(f"validating {file_name} against {definition} schema...")
             with open(file_name) as file:
-                for line_counter, line in enumerate(file):
+                for i, line in enumerate(file):
+                    line_number = i + 1
                     total_counter += 1
                     data = line.strip()
 
                     tasks.append(asyncio.create_task(
-                        self.do_validate_payload(endpoint, file_name, data, line_counter)))
+                        self.do_validate_payload(endpoint, file_name, data, line_number)))
 
                     if len(tasks) >= self.MAX_VALIDATE_TASK_QUEUE_SIZE:
                         await self.lightbeam.do_tasks(tasks, total_counter, log_status_counts=False)
@@ -235,10 +237,10 @@ async def validate_endpoint(self, endpoint):
                 num_others = self.lightbeam.num_errors - self.MAX_VALIDATION_ERRORS_TO_DISPLAY
                 if self.lightbeam.num_errors > self.MAX_VALIDATION_ERRORS_TO_DISPLAY:
                     self.logger.warn(f"... and {num_others} others!")
-                self.logger.warn(f"... VALIDATION ERRORS on {self.lightbeam.num_errors} of {line_counter} lines in {file_name}; see details above.")
+                self.logger.warn(f"... VALIDATION ERRORS on {self.lightbeam.num_errors} of {line_number} lines in {file_name}; see details above.")
 
 
-    async def do_validate_payload(self, endpoint, file_name, data, line_counter):
+    async def do_validate_payload(self, endpoint, file_name, data, line_number):
         if self.fail_fast_threshold is not None and self.lightbeam.num_errors >= self.fail_fast_threshold: return
         definition = self.get_swagger_definition_for_endpoint(endpoint)
         if "Descriptor" in endpoint:
@@ -262,7 +264,7 @@ async def do_validate_payload(self, endpoint, file_name, data, line_counter):
         try:
             payload = json.loads(data)
         except Exception as e:
-            self.log_validation_error(endpoint, file_name, line_counter, "json", f"invalid JSON {str(e).replace(' line 1','')}")
+            self.log_validation_error(endpoint, file_name, line_number, "json", f"invalid JSON {str(e).replace(' line 1','')}")
             return
 
         # check payload obeys Swagger schema
@@ -273,22 +275,22 @@ async def do_validate_payload(self, endpoint, file_name, data, line_counter):
                 e_path = [str(x) for x in list(e.path)]
                 context = ""
                 if len(e_path)>0: context = " in " + " -> ".join(e_path)
-                self.log_validation_error(endpoint, file_name, line_counter, "schema", f"{str(e.message)} {context}")
+                self.log_validation_error(endpoint, file_name, line_number, "schema", f"{str(e.message)} {context}")
                 return
 
         # check descriptor values are valid
         if "descriptors" in self.validation_methods:
             error_message = self.has_invalid_descriptor_values(payload, path="")
             if error_message != "":
-                self.log_validation_error(endpoint, file_name, line_counter, "descriptors", error_message)
+                self.log_validation_error(endpoint, file_name, line_number, "descriptors", error_message)
                 return
 
         # check natural keys are unique
         if "uniqueness" in self.validation_methods:
             params = json.dumps(util.interpolate_params(identity_params_structure, payload))
             params_hash = hashlog.get_hash(params)
             if params_hash in distinct_params:
-                self.log_validation_error(endpoint, file_name, line_counter, "uniqueness", "duplicate value(s) for natural key(s): {params}")
+                self.log_validation_error(endpoint, file_name, line_number, "uniqueness", "duplicate value(s) for natural key(s): {params}")
                 return
             else: distinct_params.append(params_hash)
 
@@ -297,7 +299,7 @@ async def do_validate_payload(self, endpoint, file_name, data, line_counter):
             self.lightbeam.api.do_oauth()
             error_message = self.has_invalid_references(payload, path="")
             if error_message != "":
-                self.log_validation_error(endpoint, file_name, line_counter, "references", error_message)
+                self.log_validation_error(endpoint, file_name, line_number, "references", error_message)
 
 
     def log_validation_error(self, endpoint, file_name, line_number, method, message):