robustness to older Ed-Fi API versions that do not report identity

Author: Tom Reitz

lightbeam/api.py

@@ -21,6 +21,7 @@ def __init__(self, lightbeam=None):
         self.lightbeam = lightbeam
         self.logger = self.lightbeam.logger
         self.config = None
+        self.reports_identity = False
 
     # prepares this API object by fetching some of its metadata and
     # setting up data and objects for further use
@@ -239,6 +240,9 @@ def load_swagger_docs(self):
                         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))
                         swagger = response.json()
+                        if '"x-Ed-Fi-isIdentity":' in response.text:
+                            self.reports_identity = True
+                            print("reports identity!!")
 
                     except Exception as e:
                         self.logger.critical(f"Unable to load {endpoint_type} Swagger from API... terminating. Check API connectivity.")
@@ -365,7 +369,7 @@ def get_identity_params_from_swagger(self, swagger, definition, prefix=""):
         schema = util.resolve_swagger_ref(swagger, definition)
         if not schema:
             self.logger.critical(f"Swagger contains neither `definitions` nor `components.schemas` - check that the Swagger is valid.")
-        
+        print(definition, schema)
         for prop in schema["properties"]:
             if prop.endswith("Reference") and "required" in schema.keys() and prop in schema['required'] and "$ref" in schema["properties"][prop].keys():
                 sub_definition = schema["properties"][prop]["$ref"]

lightbeam/delete.py

@@ -54,8 +54,11 @@ async def do_deletes(self, endpoint):
         tasks = []
 
         # determine the fields that uniquely define a record for this endpoint
-        params_structure = self.lightbeam.api.get_params_for_endpoint(endpoint, type='identity')
-
+        interpolation_type = 'required'
+        if self.lightbeam.api.reports_identity: interpolation_type = 'identity'
+        params_structure = self.lightbeam.api.get_params_for_endpoint(endpoint, type=interpolation_type)
+        print(params_structure)
+        
         # for Descriptors, we need to fetch all Descriptor values first, then we can look up the ID for deletion
         if endpoint.endswith('Descriptors'):
             self.logger.info("fetching current descriptors from endpoint {0} ...".format(endpoint))
@@ -82,6 +85,7 @@ async def do_deletes(self, endpoint):
                     # (so we can search for matching records in the API)
                     payload = json.loads(data)
                     params = util.interpolate_params(params_structure, payload)
+                    print(params)
 
                     # check if we've posted this data before
                     data_hash = hashlog.get_hash(data)