Merge pull request #62 from edanalytics/main

update `lightbeam create` branch to latest main

Author: tomreitz

CHANGELOG.md

@@ -1,3 +1,15 @@
+### v0.1.6
+<details>
+<summary>Released 2024-11-15</summary>
+
+* feature: [implement structured results file output for validate](https://github.com/edanalytics/lightbeam/pull/52)
+* bugfix: [`--results-file` required a directory prefix](https://github.com/edanalytics/lightbeam/pull/57)
+* bugfix: [`validate` of descriptor values without a hash char (`#`) failed](https://github.com/edanalytics/lightbeam/pull/53)
+* bugfix: [validate uniqueness by identity instead of required](https://github.com/edanalytics/lightbeam/pull/54)
+* bugfix: [allow deletion of students (if permitted by credential claimset and API auth strategy)](https://github.com/edanalytics/lightbeam/pull/55)
+* bugfix: [make `max_failures` truly optional](https://github.com/edanalytics/lightbeam/pull/58)
+</details>
+
 ### v0.1.5
 <details>
 <summary>Released 2024-07-23</summary>
@@ -113,4 +125,4 @@
 <summary>Released 2022-09-22</summary>
 
 * initial release
-</details>
+</details>

lightbeam/VERSION.txt

@@ -1 +1 @@
-0.1.5
+0.1.6

lightbeam/api.py

@@ -304,7 +304,7 @@ async def load_descriptors_values(self):
                 for key in v.keys():
                     if key.endswith("Id"): descriptor = key[0:-2]
                 self.descriptor_values.append([descriptor, v["namespace"], v["codeValue"], v["shortDescription"], v.get("description", "")])
-            
+
             # save
             if self.lightbeam.track_state:
                 self.logger.debug(f"saving descriptor values to {cache_file}...")
@@ -331,33 +331,47 @@ async def load_descriptors_values(self):
     # }
     # (The first element is a required attribute of the assessmentItem; the other two are required elements
     # of the required nested assessmentReference.)
-    def get_params_for_endpoint(self, endpoint):
+    def get_params_for_endpoint(self, endpoint, type='required'):
         if "Descriptor" in endpoint: swagger = self.descriptors_swagger
         else: swagger = self.resources_swagger
-        definition = util.camel_case(self.lightbeam.config["namespace"]) + "_" + util.singularize_endpoint(endpoint)
-        return self.get_required_params_from_swagger(swagger, definition)
+        definition = util.get_swagger_ref_for_endpoint(self.lightbeam.config["namespace"], swagger, endpoint)
+        if type=='required':
+            return self.get_required_params_from_swagger(swagger, definition)
+        else:
+            # descriptor endpoints all have the same structure and identity fields:
+            if "Descriptor" in endpoint:
+                return { 'namespace':'namespace', 'codeValue':'codeValue', 'shortDescription':'shortDescription'}
+            else:
+                return self.get_identity_params_from_swagger(swagger, definition)
 
     def get_required_params_from_swagger(self, swagger, definition, prefix=""):
         params = {}
-        use_definitions = False
-        if "definitions" in swagger.keys():
-            schema = swagger["definitions"][definition]
-            use_definitions = True
-        elif "components" in swagger.keys() and "schemas" in swagger["components"].keys():
-            schema = swagger["components"]["schemas"][definition]
-        else:
+        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.")
 
-        for requiredProperty in schema["required"]:
-            if "$ref" in schema["properties"][requiredProperty].keys():
-                sub_definition = schema["properties"][requiredProperty]["$ref"]
-                if use_definitions:
-                    sub_definition = sub_definition.replace("#/definitions/", "")
-                else:
-                    sub_definition = sub_definition.replace("#/components/schemas/", "")
-                sub_params = self.get_required_params_from_swagger(swagger, sub_definition, prefix=requiredProperty+".")
+        for prop in schema["required"]:
+            if "$ref" in schema["properties"][prop].keys():
+                sub_definition = schema["properties"][prop]["$ref"]
+                sub_params = self.get_required_params_from_swagger(swagger, sub_definition, prefix=prop+".")
+                for k,v in sub_params.items():
+                    params[k] = v
+            elif schema["properties"][prop]["type"]!="array":
+                params[prop] = prefix + prop
+        return params
+
+    def get_identity_params_from_swagger(self, swagger, definition, prefix=""):
+        params = {}
+        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.")
+        
+        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"]
+                sub_params = self.get_identity_params_from_swagger(swagger, sub_definition, prefix=prop+".")
                 for k,v in sub_params.items():
                     params[k] = v
-            elif schema["properties"][requiredProperty]["type"]!="array":
-                params[requiredProperty] = prefix + requiredProperty
+            elif "type" in schema["properties"][prop].keys() and schema["properties"][prop]["type"]!="array" and "x-Ed-Fi-isIdentity" in schema["properties"][prop].keys():
+                params[prop] = prefix + prop
         return params

lightbeam/delete.py

@@ -36,12 +36,6 @@ def delete(self):
         self.lightbeam.confirm_delete(endpoints)
 
         for endpoint in endpoints:
-            # it doesn't seem possible to delete students once you've sent them
-            # (I think because other entities may have referenced them in the meantime)
-            if endpoint=='students':
-                self.logger.warn("data for {0} endpoint cannot be deleted (this is an Ed-Fi limitation); skipping".format(endpoint))
-                continue
-
             asyncio.run(self.do_deletes(endpoint))
             self.logger.info("finished processing endpoint {0}!".format(endpoint))
             self.logger.info("  (final status counts: {0})".format(self.lightbeam.status_counts))

lightbeam/lightbeam.py

@@ -82,7 +82,7 @@ def __init__(self, config_file, logger=None, selector="*", exclude="", keep_keys
         self.truncator = Truncator(self)
         self.api = EdFiAPI(self)
         self.token_version = 0        
-        self.results_file = results_file
+        self.results_file = os.path.abspath(results_file) if results_file else None
         self.start_timestamp = datetime.now()
 
         # load params and/or env vars for config YAML interpolation
@@ -168,7 +168,8 @@ def write_structured_output(self, command):
                 # failures.line_numbers are split each on their own line; here we remove those line breaks
                 content = re.sub(r'"line_numbers": \[(\d|,|\s|\n)*\]', self.replace_linebreaks, content)
                 fp.write(content)
-        self.logger.info(f"results written to {self.results_file}")
+    
+            self.logger.info(f"results written to {self.results_file}")
 
 
     def load_config_file(self) -> dict:

lightbeam/truncate.py

@@ -28,12 +28,6 @@ def truncate(self):
         self.lightbeam.confirm_truncate(endpoints)
 
         for endpoint in endpoints:
-            # it doesn't seem possible to delete students once you've sent them
-            # (I think because other entities may have referenced them in the meantime)
-            # if endpoint=='students':
-            #     self.logger.warn("data for {0} endpoint cannot be deleted (this is an Ed-Fi limitation); skipping".format(endpoint))
-            #     continue
-
             asyncio.run(self.do_truncates(endpoint))
             self.logger.info("finished processing endpoint {0}!".format(endpoint))
             self.logger.info("  (final status counts: {0})".format(self.lightbeam.status_counts))

lightbeam/util.py

@@ -79,4 +79,18 @@ def keys_match(key, wildcard_key):
     if key==wildcard_key: return True
     if wildcard_key.startswith("*") and key.endswith(wildcard_key.lstrip("*")): return True
     if wildcard_key.endswith("*") and key.startswith(wildcard_key.rstrip("*")): return True
-    return False
+    return False
+
+def get_swagger_ref_for_endpoint(namespace, swagger, endpoint):
+    if "definitions" in swagger.keys():
+        return "#/definitions/" + camel_case(namespace) + "_" + singularize_endpoint(endpoint)
+    elif "components" in swagger.keys() and "schemas" in swagger["components"].keys():
+        return "#/components/schemas/" + camel_case(namespace) + "_" + singularize_endpoint(endpoint)
+
+def resolve_swagger_ref(swagger, ref):
+    if "definitions" in swagger.keys():
+        definition = ref.replace("#/definitions/", "")
+        return swagger["definitions"][definition]
+    elif "components" in swagger.keys() and "schemas" in swagger["components"].keys():
+        definition = ref.replace("#/components/schemas/", "")
+        return swagger["components"]["schemas"][definition]

lightbeam/validate.py

@@ -14,7 +14,6 @@ class Validator:
     MAX_VALIDATION_ERRORS_TO_DISPLAY = 10
     MAX_VALIDATE_TASK_QUEUE_SIZE = 100
     DEFAULT_VALIDATION_METHODS = ["schema", "descriptors", "uniqueness"]
-    DEFAULT_FAIL_FAST_THRESHOLD = 10
 
     EDFI_GENERICS_TO_RESOURCES_MAPPING = {
         "educationOrganizations": ["localEducationAgencies", "stateEducationAgencies", "schools"],
@@ -35,7 +34,7 @@ def __init__(self, lightbeam=None):
     def validate(self):
 
         # The below should go in __init__(), but rely on lightbeam.config which is not yet available there.
-        self.fail_fast_threshold = self.lightbeam.config.get("validate",{}).get("references",{}).get("max_failures", self.DEFAULT_FAIL_FAST_THRESHOLD)
+        self.fail_fast_threshold = self.lightbeam.config.get("validate",{}).get("references",{}).get("max_failures", None)
         self.validation_methods = self.lightbeam.config.get("validate",{}).get("methods",self.DEFAULT_VALIDATION_METHODS)
         if type(self.validation_methods)==str and (self.validation_methods=="*" or self.validation_methods.lower()=='all'):
             self.validation_methods = self.DEFAULT_VALIDATION_METHODS
@@ -219,7 +218,7 @@ async def validate_endpoint(self, endpoint):
                     self.lightbeam.metadata["resources"][endpoint]["records_failed"] = self.lightbeam.num_errors
 
                     # implement "fail fast" feature:
-                    if self.lightbeam.num_errors >= self.fail_fast_threshold:
+                    if self.fail_fast_threshold is not None and self.lightbeam.num_errors >= self.fail_fast_threshold:
                         self.lightbeam.shutdown("validate")
                         self.logger.critical(f"... STOPPING; found {self.lightbeam.num_errors} >= validate.references.max_failures={self.fail_fast_threshold} VALIDATION ERRORS.")
                         break
@@ -235,7 +234,7 @@ async def validate_endpoint(self, endpoint):
 
 
     async def do_validate_payload(self, endpoint, file_name, data, line_counter):
-        if self.lightbeam.num_errors >= self.fail_fast_threshold: return
+        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:
             swagger = self.lightbeam.api.descriptors_swagger
@@ -251,7 +250,7 @@ async def do_validate_payload(self, endpoint, file_name, data, line_counter):
 
         resolver = RefResolver("test", swagger, swagger)
         validator = Draft4Validator(resource_schema, resolver=resolver)
-        params_structure = self.lightbeam.api.get_params_for_endpoint(endpoint)
+        identity_params_structure = self.lightbeam.api.get_params_for_endpoint(endpoint, type='identity')
         distinct_params = []
 
         # check payload is valid JSON
@@ -281,7 +280,7 @@ async def do_validate_payload(self, endpoint, file_name, data, line_counter):
 
         # check natural keys are unique
         if "uniqueness" in self.validation_methods:
-            params = json.dumps(util.interpolate_params(params_structure, payload))
+            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}")
@@ -344,6 +343,8 @@ def has_invalid_descriptor_values(self, payload, path=""):
                     value = self.has_invalid_descriptor_values(payload[k][i], path+("." if path!="" else "")+k+"["+str(i)+"]")
                     if value!="": return value
             elif isinstance(payload[k], str) and k.endswith("Descriptor"):
+                if "#" not in payload[k]:
+                    return payload[k] + f" is not a valid descriptor value for {k}" + (" (at " + path + ")" if path!="" else "") + "; format should be like `uri://namespace.org/SomeDescriptor#SomeValue`"
                 namespace = payload[k].split("#")[0]
                 codeValue = payload[k].split("#")[1]
                 # check if it's a local descriptor: