Merge branch 'main' into fix/command_list

Author: tomreitz

CHANGELOG.md

@@ -2,6 +2,7 @@
 <details>
 
 * bugfix: [Ensure command list in help menu and log output is always consistent](https://github.com/edanalytics/lightbeam/pull/27)
+* bugfix: Fix how hashlog entries are removed during `lightbeam delete`
 </details>
 
 ### v0.1.2

README.md

@@ -43,6 +43,10 @@ data_dir: ./
 namespace: ed-fi
 edfi_api:
   base_url: https://api.schooldistrict.org/v5.3/api
+  oauth_url: https://api.schooldistrict.org/v5.3/api/oauth/token 
+  dependencies_url: https://api.schooldistrict.org/v5.3/api/metadata/data/v3/2024/dependencies
+  descriptors_swagger_url: https://api.schooldistrict.org/v5.3/api/metadata/data/v3/2024/descriptors/swagger.json
+  resources_swagger_url: https://api.schooldistrict.org/v5.3/api/metadata/data/v3/2024/resources/swagger.json
   version: 3
   mode: year_specific
   year: 2021
@@ -68,6 +72,11 @@ show_stacktrace: True
 * (optional) Specify the `namespace` to use when accessing the Ed-Fi API. The default is `ed-fi` but others include `tpdm` or custom values. To send data to multiple namespaces, you must use a YAML configuration file and `lightbeam send` for each.
 * Specify the details of the `edfi_api` to which to connect including
   * (optional) The `base_url` which serves a JSON object specifying the paths to data endpoints, Swagger, and dependencies. The default is `https://localhost/api` (the address of an Ed-Fi API [running locally in Docker](https://techdocs.ed-fi.org/display/EDFITOOLS/Docker+Deployment)), but the location varies depending on how Ed-Fi is deployed.
+  * If the metadata for a particular API is not located in the "default" location (at the root of the base_url), then ALL the following urls should be explicitly specified.  These can normally be left blank, unless you are encountering errors indicating that the metadata files cannot be found (such as "Could not parse response from [base_url]").
+    * (optional) `oauth_url` (usually [base_url]/oauth/token)
+    * (optional) `dependencies_url` (usually [base_url]/metadata/data/v3/dependencies)
+    * (optional) `descriptors_swagger_url` (usually [base_url]/metadata/data/v3/descriptors/swagger.json)
+    * (optional) `resources_swagger_url` (usually [base_url]/metadata/data/v3/resources/swagger.json)
   * The `version` as one of `3` or `2` (`2` is currently unsupported).
   * (optional) The `mode` as one of `shared_instance`, `sandbox`, `district_specific`, `year_specific`, or `instance_year_specific`.
   * (required if `mode` is `year_specific` or `instance_year_specific`) The `year` used to build the resource URL. The default is the current year.

lightbeam/__main__.py

@@ -148,7 +148,7 @@ def main(argv=None):
         logger=logger,
         selector=args.selector or "*",
         exclude=args.exclude or "",
-        keep_keys=args.keep_keys or "",
+        keep_keys=args.keep_keys or "*",
         drop_keys=args.drop_keys or "",
         query=args.query or "{}",
         params=args.params,

lightbeam/api.py

@@ -35,15 +35,22 @@ def prepare(self):
         except Exception as e:
             self.logger.critical("could not connect to {0} ({1})".format(self.config["base_url"], str(e)))
 
-        try:
-            api_base = api_base.json()
-        except Exception as e:
-            self.logger.critical("could not parse response from {0} ({1})".format(self.config["base_url"], str(e)))
-
-        self.config["oauth_url"] = api_base["urls"]["oauth"]
-        self.config["dependencies_url"] = api_base["urls"]["dependencies"]
+        # Data URL doesn't rely on metadata connection
         self.config["data_url"] = self.get_data_url()
-        self.config["open_api_metadata_url"] = api_base["urls"]["openApiMetadata"]
+
+        # If ALL urls are set in config (probably from source/destination file),
+        # then they don't need to be pulled from api metadata, so this section can be skipped.
+        # This is most common if the api metadata json files are not in the "default" location.
+        # Otherwise, pull urls from api metadata.
+        if self.config["oauth_url"] == "" or self.config["dependencies_url"]== "":
+            try:
+                api_base = api_base.json()
+            except Exception as e:
+                self.logger.critical("could not parse response from {0} ({1})".format(self.config["base_url"], str(e)))
+
+            self.config["oauth_url"] = api_base["urls"]["oauth"]
+            self.config["dependencies_url"] = api_base["urls"]["dependencies"]
+            self.config["open_api_metadata_url"] = api_base["urls"]["openApiMetadata"]
 
         # load all endpoints in dependency-order
         all_endpoints = self.get_sorted_endpoints()
@@ -160,17 +167,33 @@ def get_sorted_endpoints(self):
 
     # Loads the Swagger JSON from the Ed-Fi API
     def load_swagger_docs(self):
-        # grab Descriptors and Resources swagger URLs
-        try:
-            self.logger.debug("fetching swagger docs...")
-            response = requests.get(self.config["open_api_metadata_url"],
-                                    verify=self.lightbeam.config["connection"]["verify_ssl"])
-            if not response.ok:
-                raise Exception("OpenAPI metadata URL returned status {0} ({1})".format(response.status_code, (response.content[:75] + "...") if len(response.content)>75 else response.content))
-            openapi = response.json()
 
-        except Exception as e:
-            self.logger.critical("Unable to load Swagger docs from API... terminating. Check API connectivity.")
+        # If the metadata URL is set (pulled from root metadata file earlier), then pull endpoint urls from metadata
+        if "open_api_metadata_url" in self.config.keys() and self.config["open_api_metadata_url"] != "":
+            # grab Descriptors and Resources swagger URLs
+            try:
+                self.logger.debug("fetching swagger docs...")
+                response = requests.get(self.config["open_api_metadata_url"],
+                                        verify=self.lightbeam.config["connection"]["verify_ssl"])
+                if not response.ok:
+                    raise Exception("OpenAPI metadata URL returned status {0} ({1})".format(response.status_code, (response.content[:75] + "...") if len(response.content)>75 else response.content))
+                openapi = response.json()
+
+            except Exception as e:
+                self.logger.critical("Unable to load Swagger docs from API... terminating. Check API connectivity.")
+        
+        # If metadata URL is not found, set endpoint URLs from config file
+        else:
+            openapi = [
+                {
+                    "name": "descriptors",
+                    "endpointUri": self.config["descriptors_swagger_url"],
+                }, 
+                {
+                    "name": "resources",
+                    "endpointUri": self.config["resources_swagger_url"],
+                }
+            ]
 
         # load (or re-use cached) Descriptors and Resources swagger
         self.descriptors_swagger = None
@@ -187,8 +210,8 @@ def load_swagger_docs(self):
             if endpoint_type=="descriptors" or endpoint_type=="resources":
                 swagger_url = endpoint["endpointUri"]
                 if self.lightbeam.track_state:
-                    hash = hashlog.get_hash_string(swagger_url)
-                    file = os.path.join(cache_dir, f"swagger-{endpoint_type}-{hash}.json")
+                    url_hash = hashlog.get_hash_string(swagger_url)
+                    file = os.path.join(cache_dir, f"swagger-{endpoint_type}-{url_hash}.json")
                 if (
                     self.lightbeam.track_state  # we have a state_dir in which to store
                     and not self.lightbeam.wipe # we aren't clearing the cache
@@ -234,8 +257,8 @@ async def load_descriptors_values(self):
                 os.mkdir(cache_dir)
 
             # check for cached descriptor values
-            hash = hashlog.get_hash_string(self.config["base_url"])
-            cache_file = os.path.join(cache_dir, f"descriptor-values-{hash}.csv")
+            url_hash = hashlog.get_hash_string(self.config["base_url"])
+            cache_file = os.path.join(cache_dir, f"descriptor-values-{url_hash}.csv")
 
         self.lightbeam.reset_counters()
         if (
@@ -256,8 +279,12 @@ async def load_descriptors_values(self):
             # load descriptor values from API
             selector_backup = self.lightbeam.selector
             exclude_backup = self.lightbeam.exclude
+            keep_keys_backup = self.lightbeam.keep_keys
+            drop_keys_backup = self.lightbeam.drop_keys
             self.lightbeam.selector = "*Descriptors"
             self.lightbeam.exclude = ""
+            self.lightbeam.keep_keys = "*"
+            self.lightbeam.drop_keys = ""
             self.logger.debug(f"fetching descriptor values...")
             all_endpoints = self.get_sorted_endpoints()
             self.lightbeam.endpoints = self.apply_filters(all_endpoints)
@@ -281,6 +308,8 @@ async def load_descriptors_values(self):
             self.lightbeam.results = []
             self.lightbeam.selector = selector_backup
             self.lightbeam.exclude = exclude_backup
+            self.lightbeam.keep_keys = keep_keys_backup
+            self.lightbeam.drop_keys = drop_keys_backup
             self.prepare()
 
 

lightbeam/delete.py

@@ -91,16 +91,13 @@ async def do_deletes(self, endpoint):
                     params = util.interpolate_params(params_structure, data)
 
                     # check if we've posted this data before
-                    hash = hashlog.get_hash(data)
-                    if self.lightbeam.track_state and hash in self.hashlog_data.keys():
+                    data_hash = hashlog.get_hash(data)
+                    if self.lightbeam.track_state and data_hash in self.hashlog_data.keys():
                         # check if the last post meets criteria for a delete
-                        if self.lightbeam.meets_process_criteria(self.hashlog_data[hash]):
+                        if self.lightbeam.meets_process_criteria(self.hashlog_data[data_hash]):
                             # yes, we need to delete it; append to task queue
                             tasks.append(asyncio.create_task(
-                                self.do_delete(endpoint, file_name, params, counter)))
-                            
-                            # remove the payload from the hashlog
-                            del self.hashlog_data[hash]
+                                self.do_delete(endpoint, file_name, params, counter, data_hash)))
                         else:
                             # no, do not delete
                             self.lightbeam.num_skipped += 1
@@ -128,7 +125,7 @@ async def do_deletes(self, endpoint):
             hashlog.save(hashlog_file, self.hashlog_data)
 
     # Deletes a single payload for a single endpoint
-    async def do_delete(self, endpoint, file_name, params, line):
+    async def do_delete(self, endpoint, file_name, params, line, data_hash=None):
         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:
@@ -163,7 +160,7 @@ async def do_delete(self, endpoint, file_name, params, line):
                                 if type(j)==list and len(j)==1:
                                     the_id = j[0]['id']
                                     # now we can delete by `id`
-                                    await self.do_delete_id(endpoint, the_id, file_name, line)
+                                    await self.do_delete_id(endpoint, the_id, file_name, line, data_hash)
                                     break
 
                                 elif type(j)==list and len(j)==0: skip_reason = "payload not found in API"
@@ -195,7 +192,7 @@ async def do_delete(self, endpoint, file_name, params, line):
                 self.logger.error("  (at line {0} of {1}; ID: {2} )".format(line, file_name, id))
                 break
 
-    async def do_delete_id(self, endpoint, id, file_name=None, line=None):
+    async def do_delete_id(self, endpoint, id, file_name=None, line=None, data_hash=None):
         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:
@@ -209,12 +206,15 @@ async def do_delete_id(self, endpoint, id, file_name=None, line=None):
                     if status!=401:
                         self.lightbeam.num_finished += 1
                         self.lightbeam.increment_status_counts(status)
-                        if self.lightbeam.track_state:
-                            del self.hashlog_data[hash]
                         if status not in [ 204 ]:
                             message = str(status) + ": " + util.linearize(body)
                             self.lightbeam.increment_status_reason(message)
                             self.lightbeam.num_errors += 1
+                        else:
+                            if self.lightbeam.track_state and data_hash is not None:
+                                # if we're certain delete was successful, remove this
+                                # line of data from internal tracking
+                                del self.hashlog_data[data_hash]
                         break # (out of while loop)
                     else:
                         # this could be broken out to a separate function call,
@@ -233,4 +233,4 @@ async def do_delete_id(self, endpoint, id, file_name=None, line=None):
                 self.logger.exception(e, exc_info=self.lightbeam.config["show_stacktrace"])
                 if line and file_name:
                     self.logger.error("  (at line {0} of {1}; ID: {2} )".format(line, file_name, id))
-                break
+                break

lightbeam/fetch.py

@@ -89,8 +89,9 @@ async def get_endpoint_records(self, endpoint, limit, offset, file_handle=None):
                                 payload_keys = list(values[0].keys())
                                 final_keys = util.apply_selections(payload_keys, self.lightbeam.keep_keys, self.lightbeam.drop_keys)
                                 do_key_filtering = len(payload_keys) != len(final_keys)
+
                                 for v in values:
-                                    if do_key_filtering: row = {k: v[k] for k in final_keys}
+                                    if do_key_filtering: row = {k: v.get(k, None) for k in final_keys} #v.get() to account for missing keys
                                     else: row = v
                                     if file_handle: file_handle.write(json.dumps(row)+"\n")
                                     else: self.lightbeam.results.append(row)

lightbeam/lightbeam.py

@@ -24,6 +24,10 @@ class Lightbeam:
         "namespace": "ed-fi",
         "edfi_api": {
             "base_url": "",
+            "oauth_url": "", 
+            "dependencies_url": "",
+            "descriptors_swagger_url": "",
+            "resources_swagger_url": "",
             "version": 3,
             "mode": "year_specific",
             "year": datetime.today().year,
@@ -50,7 +54,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=""):
         self.config_file = config_file
         self.logger = logger
         self.errors = 0