Merge branch 'main' into feature/validate_fixes_and_uniqueness_in_array_elements

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/__main__.py

@@ -108,6 +108,11 @@ def main(argv=None):
         type=str,
         help='produces a JSON output file with structured information about run results'
     )
+    parser.add_argument("--set",
+        type=str,
+        nargs="*",
+        help='overrides a setting in the config YAML; example: --set fetch.page_size 1000'
+    )
 
     defaults = { "selector":"*", "params": "", "older_than": "", "newer_than": "", "resend_status_codes": "", "results_file": "" }
     parser.set_defaults(**defaults)
@@ -143,6 +148,12 @@ def main(argv=None):
 
     if not args.config_file:
         logger.error("config file not specified with `-c` flag, and no default {" + ", ".join(DEFAULT_CONFIG_FILES) + "} found")
+    
+    if args.set and len(args.set)%2 != 0: # odd number of overrides
+        logger.error("overrides specified with --set must be followed by an even number of strings (key value key value ...)")
+    overrides = None
+    if args.set:
+        overrides = dict(zip(args.set[::2], args.set[1::2]))
 
     lb = Lightbeam(
         config_file=args.config_file,
@@ -158,7 +169,8 @@ def main(argv=None):
         older_than=args.older_than,
         newer_than=args.newer_than,
         resend_status_codes=args.resend_status_codes,
-        results_file=args.results_file
+        results_file=args.results_file,
+        overrides=overrides,
         )
     try:
         logger.info("starting...")

lightbeam/lightbeam.py

@@ -55,7 +55,7 @@ class Lightbeam:
     MAX_STATUS_REASONS_TO_DISPLAY = 10
     DATA_FILE_EXTENSIONS = ['json', 'jsonl', 'ndjson']
 
-    def __init__(self, config_file, logger=None, selector="*", exclude="", keep_keys="*", drop_keys="", query="{}", params="", wipe=False, force=False, older_than="", newer_than="", resend_status_codes="", results_file=""):
+    def __init__(self, config_file, logger=None, selector="*", exclude="", keep_keys="*", drop_keys="", query="{}", params="", wipe=False, force=False, older_than="", newer_than="", resend_status_codes="", results_file="", overrides={}):
         self.config_file = config_file
         self.logger = logger
         self.errors = 0
@@ -82,12 +82,16 @@ def __init__(self, config_file, logger=None, selector="*", exclude="", keep_keys
         self.token_version = 0        
         self.results_file = os.path.abspath(results_file) if results_file else None
         self.start_timestamp = datetime.now()
+        self.overrides = overrides
 
         # load params and/or env vars for config YAML interpolation
         self.params = json.loads(params) if params else {}
         user_config = self.load_config_file()
-        
         self.config = util.merge_dicts(user_config, self.config_defaults)
+        
+        # inject overrides into config
+        if self.overrides: self.inject_cli_overrides()
+
         if "state_dir" in self.config:
             self.track_state = True
             self.config["state_dir"] = os.path.expanduser(self.config["state_dir"])
@@ -129,6 +133,39 @@ def __init__(self, config_file, logger=None, selector="*", exclude="", keep_keys
             "resources": {}
         }
 
+    def inject_cli_overrides(self):
+        # parse self.overrides into configs:
+        for key, value in self.overrides.items():
+            self.config = Lightbeam.set_path(self.config, key, value)
+
+    @staticmethod
+    def set_path(my_dict, path, value):
+        path_pieces = path.split(".")
+        current = my_dict
+        for path_piece in path_pieces[:-1]:
+            if path_piece not in current.keys():
+                current[path_piece] = {}
+            current = current[path_piece]
+        current[path_pieces[-1]] = Lightbeam.autocast(value)
+        return my_dict
+    
+    @staticmethod
+    def autocast(value):
+        if value.lower() in ['true', 'yes', 'on', 't', 'y']:
+            return True
+        elif value.lower() in ['false', 'no', 'off', 'f', 'n']:
+            return False
+        elif '.' in value:
+            try:
+                return float(value)
+            except ValueError:
+                return value
+        else:
+            try:
+                return int(value)
+            except ValueError:
+                return value
+    
     # this is intended to be called before any CRITICAL errors;
     # any cleanup tasks should go here:
     def shutdown(self, method):
@@ -255,9 +292,10 @@ def get_endpoints_with_data(self, filter_endpoints=None):
                         if os.path.isfile(sub_dir_item_path):
                             filename = os.path.basename(sub_dir_item)
                             extension = filename.rsplit(".", 1)[-1]
+                            filename_without_extension = filename.rsplit(".", 1)[0]
                             if (
                                 extension in self.DATA_FILE_EXTENSIONS # valid file extension
-                                and filename_without_extension in self.all_endpoints # valid endpoint
+                                and data_dir_item in self.all_endpoints # valid endpoint
                                 and data_dir_item in filter_endpoints # selected endpoint
                             ):
                                 has_data_file = True

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)
@@ -217,12 +218,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)
@@ -243,6 +245,11 @@ async def validate_endpoint(self, endpoint):
 
             if len(tasks)>0: await self.lightbeam.do_tasks(tasks, total_counter, log_status_counts=False)
 
+            # update metadata counts
+            self.lightbeam.metadata["resources"][endpoint]["records_processed"] = total_counter
+            self.lightbeam.metadata["resources"][endpoint]["records_skipped"] = self.lightbeam.num_skipped
+            self.lightbeam.metadata["resources"][endpoint]["records_failed"] = self.lightbeam.num_errors
+            
             if self.lightbeam.num_errors==0: self.logger.info(f"... all lines validate ok!")
             else:
                 num_others = self.lightbeam.num_errors - self.MAX_VALIDATION_ERRORS_TO_DISPLAY
@@ -257,14 +264,14 @@ async def validate_endpoint(self, endpoint):
         self.schema_validator = None
 
 
-    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
 
         # check payload is valid JSON
         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
@@ -275,14 +282,14 @@ 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
@@ -298,7 +305,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):