import jsonschema import json #import fastjsonschema class BehaverseJsonValidator(): """ Loads Behaverse json file from local drive, aggregate json lines into a single array, and validate the content and structure of the events. """ def __init__(self, json_schema_path: str = None): if json_schema_path is not None: with open(json_schema_path) as f: self.schema = json.load(f) @staticmethod def __join_json_lines(src_path): with open(src_path) as f: return f"[{','.join(f.readlines())}]" def convert(self, src_path, dest_path): with open(dest_path, "w") as f: f.write(self.__join_json_lines(src_path)) def validate_json(self, content): """ Validates a json object with respect to a predefined schema (json-schema-v7). If json contains errornous content, a validation error will be raised. """ assert(self.schema is not None) return jsonschema.validate(content, self.schema) def validate_json_file(self, json_path: str): """ Loads and validates a json file. Respective validation error will be raised in case of invalid structure. """ with open(json_path) as f: content = json.load(f) return self.validate_json(content) if __name__ == "__main__": """ Example code showing how to use this class. """ original_json_file = "data/samples/nback_demo.json" valid_json_file = "data/samples/nback_demo_v.json" schema_file = "behaverse/validators/nback.schema.json" valid_json_file = "behaverse/validators/test.json" schema_file = "behaverse/validators/test.schema.json" BehaverseJsonValidator(json_schema_path=schema_file).validate_json_file(valid_json_file) # load and concat json lines #JsonAggregator().convert(original_json_file, valid_json_file) # load and validate json lines #BehaverseJsonValidator(json_schema_path=schema_file).validate_json_file(valid_json_file) print("Good news everyone! Behaverse JSON file is valid!")