123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596 |
- """Functions that help us generate and use info.json files.
- """
- import json
- from collections.abc import Mapping
- from functools import lru_cache
- from pathlib import Path
- import hjson
- import jsonschema
- from milc import cli
- def json_load(json_file):
- """Load a json file from disk.
- Note: file must be a Path object.
- """
- try:
- # Get the IO Stream for Path objects
- # Not necessary if the data is provided via stdin
- if isinstance(json_file, Path):
- json_file = json_file.open(encoding='utf-8')
- return hjson.load(json_file)
- except (json.decoder.JSONDecodeError, hjson.HjsonDecodeError) as e:
- cli.log.error('Invalid JSON encountered attempting to load {fg_cyan}%s{fg_reset}:\n\t{fg_red}%s', json_file, e)
- exit(1)
- except Exception as e:
- cli.log.error('Unknown error attempting to load {fg_cyan}%s{fg_reset}:\n\t{fg_red}%s', json_file, e)
- exit(1)
- @lru_cache(maxsize=0)
- def load_jsonschema(schema_name):
- """Read a jsonschema file from disk.
- """
- if Path(schema_name).exists():
- return json_load(schema_name)
- schema_path = Path(f'data/schemas/{schema_name}.jsonschema')
- if not schema_path.exists():
- schema_path = Path('data/schemas/false.jsonschema')
- return json_load(schema_path)
- @lru_cache(maxsize=0)
- def compile_schema_store():
- """Compile all our schemas into a schema store.
- """
- schema_store = {}
- for schema_file in Path('data/schemas').glob('*.jsonschema'):
- schema_data = load_jsonschema(schema_file)
- if not isinstance(schema_data, dict):
- cli.log.debug('Skipping schema file %s', schema_file)
- continue
- schema_store[schema_data['$id']] = schema_data
- return schema_store
- @lru_cache(maxsize=0)
- def create_validator(schema):
- """Creates a validator for the given schema id.
- """
- schema_store = compile_schema_store()
- resolver = jsonschema.RefResolver.from_schema(schema_store[schema], store=schema_store)
- # TODO: Remove this after the jsonschema>=4 requirement had time to reach users
- try:
- return jsonschema.Draft202012Validator(schema_store[schema], resolver=resolver).validate
- except AttributeError:
- return jsonschema.Draft7Validator(schema_store[schema], resolver=resolver).validate
- def validate(data, schema):
- """Validates data against a schema.
- """
- validator = create_validator(schema)
- return validator(data)
- def deep_update(origdict, newdict):
- """Update a dictionary in place, recursing to do a depth-first deep copy.
- """
- for key, value in newdict.items():
- if isinstance(value, Mapping):
- origdict[key] = deep_update(origdict.get(key, {}), value)
- else:
- origdict[key] = value
- return origdict
|