json_schema.py 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596
  1. """Functions that help us generate and use info.json files.
  2. """
  3. import json
  4. from collections.abc import Mapping
  5. from functools import lru_cache
  6. from pathlib import Path
  7. import hjson
  8. import jsonschema
  9. from milc import cli
  10. def json_load(json_file):
  11. """Load a json file from disk.
  12. Note: file must be a Path object.
  13. """
  14. try:
  15. # Get the IO Stream for Path objects
  16. # Not necessary if the data is provided via stdin
  17. if isinstance(json_file, Path):
  18. json_file = json_file.open(encoding='utf-8')
  19. return hjson.load(json_file)
  20. except (json.decoder.JSONDecodeError, hjson.HjsonDecodeError) as e:
  21. cli.log.error('Invalid JSON encountered attempting to load {fg_cyan}%s{fg_reset}:\n\t{fg_red}%s', json_file, e)
  22. exit(1)
  23. except Exception as e:
  24. cli.log.error('Unknown error attempting to load {fg_cyan}%s{fg_reset}:\n\t{fg_red}%s', json_file, e)
  25. exit(1)
  26. @lru_cache(maxsize=0)
  27. def load_jsonschema(schema_name):
  28. """Read a jsonschema file from disk.
  29. """
  30. if Path(schema_name).exists():
  31. return json_load(schema_name)
  32. schema_path = Path(f'data/schemas/{schema_name}.jsonschema')
  33. if not schema_path.exists():
  34. schema_path = Path('data/schemas/false.jsonschema')
  35. return json_load(schema_path)
  36. @lru_cache(maxsize=0)
  37. def compile_schema_store():
  38. """Compile all our schemas into a schema store.
  39. """
  40. schema_store = {}
  41. for schema_file in Path('data/schemas').glob('*.jsonschema'):
  42. schema_data = load_jsonschema(schema_file)
  43. if not isinstance(schema_data, dict):
  44. cli.log.debug('Skipping schema file %s', schema_file)
  45. continue
  46. schema_store[schema_data['$id']] = schema_data
  47. return schema_store
  48. @lru_cache(maxsize=0)
  49. def create_validator(schema):
  50. """Creates a validator for the given schema id.
  51. """
  52. schema_store = compile_schema_store()
  53. resolver = jsonschema.RefResolver.from_schema(schema_store[schema], store=schema_store)
  54. # TODO: Remove this after the jsonschema>=4 requirement had time to reach users
  55. try:
  56. return jsonschema.Draft202012Validator(schema_store[schema], resolver=resolver).validate
  57. except AttributeError:
  58. return jsonschema.Draft7Validator(schema_store[schema], resolver=resolver).validate
  59. def validate(data, schema):
  60. """Validates data against a schema.
  61. """
  62. validator = create_validator(schema)
  63. return validator(data)
  64. def deep_update(origdict, newdict):
  65. """Update a dictionary in place, recursing to do a depth-first deep copy.
  66. """
  67. for key, value in newdict.items():
  68. if isinstance(value, Mapping):
  69. origdict[key] = deep_update(origdict.get(key, {}), value)
  70. else:
  71. origdict[key] = value
  72. return origdict