wrpylib/json_tools.py

   1 from typing import Union, Dict, List
   2
   3
   4 JsonTypes = Union[Dict, List, str, int, float, bool, None]
   5
   6
   7 class ValidationError(ValueError):
   8     pass
   9
  10
  11 def _fmt_path(path: List) -> str:
  12     return f'schema[{"][".join(map(repr, path))}]'
  13
  14
  15 def _resolve_ref_not_recursive(sub_schema: JsonTypes, schema: JsonTypes) -> JsonTypes:
  16     """In case the sub_schema is a dict and has direct "$ref" keys,
  17     it is replaced by a dict where the "$ref" key is replaced by the corresponding definition in schema.
  18     Nested $ref keys are not resolved.
  19     Recursive $ref keys are not resolved.
  20
  21     :param sub_schema: JSON sub-schema where a "$ref" key is possible replaced.
  22         The value of "$ref" could be e.g. "#/definitions/position"
  23     :param schema: JSON root schema containing definitions for the keys.
  24     :raise ValidationError: In case a "$ref" could not be resolved.
  25     """
  26     if not isinstance(sub_schema, dict) or '$ref' not in sub_schema:
  27         return sub_schema
  28     ref = sub_schema['$ref']
  29     if not isinstance(ref, str):
  30         raise ValidationError(f'Type of reference {ref} is not string.')
  31     path = ref.split('/')
  32     if len(path) == 0 or path[0] != '#':
  33         raise ValidationError(f'Unsupported reference {ref}.')
  34     ref_schema = schema
  35     for p in path[1:]:
  36         if not isinstance(ref_schema, dict) or p not in ref_schema:
  37             raise ValidationError(f'Reference path {ref} not found in schema.')
  38         ref_schema = ref_schema[p]
  39     if not isinstance(ref_schema, dict):
  40         raise ValidationError(f'Reference path {ref} is no dict.')
  41     sub_schema = sub_schema.copy()
  42     del sub_schema['$ref']
  43     resolved_schema = ref_schema.copy()
  44     resolved_schema.update(sub_schema)
  45     return resolved_schema
  46
  47
  48 def _resolve_ref(sub_schema: JsonTypes, schema: JsonTypes) -> JsonTypes:
  49     """Same as `_resolve_ref_not_recursive` but recursively resolves $ref keys.
  50     However, does not resolve nested $ref keys.
  51     """
  52     resolved_schema = sub_schema
  53     while isinstance(resolved_schema, dict) and '$ref' in resolved_schema:
  54         resolved_schema = _resolve_ref_not_recursive(resolved_schema, schema)
  55     return resolved_schema
  56
  57
  58 def _order_json_keys_string(sub_value: JsonTypes, sub_schema: JsonTypes, schema: JsonTypes, path: List) -> str:
  59     if not isinstance(sub_value, str):
  60         raise ValidationError(f'Type of {_fmt_path(path)} needs to be string (Python str).')
  61     return sub_value
  62
  63
  64 def _order_json_keys_number(sub_value: JsonTypes, sub_schema: JsonTypes,
  65                             schema: JsonTypes, path: List) -> Union[int, float]:
  66     if not isinstance(sub_value, (int, float)) or isinstance(sub_value, bool):
  67         raise ValidationError(f'Type of {_fmt_path(path)} needs to be number (Python int or float).')
  68     return sub_value
  69
  70
  71 def _order_json_keys_object(sub_value: JsonTypes, sub_schema: JsonTypes, schema: JsonTypes, path: List) -> Dict:
  72     if not isinstance(sub_value, dict):
  73         raise ValidationError(f'Type of {_fmt_path(path)} needs to be object (Python dict).')
  74     v = sub_value.copy()
  75     p = sub_schema.get('properties', {})
  76     result = {}
  77     for key in p:
  78         if key in v:
  79             result[key] = _order_json_keys(v.pop(key), p[key], schema, path + [key])
  80         else:
  81             if key in sub_schema.get('required', []):
  82                 raise ValidationError(f'Required key "{key}" not present ({_fmt_path(path)}).')
  83     if len(v) > 0:
  84         if sub_schema.get('additionalProperties', True):
  85             # strictly speaking additionalProperties could be more complicated than boolean
  86             result.update(v)
  87         else:
  88             raise ValidationError(f'Keys not allowed in {_fmt_path(path)}: {", ".join(v)}')
  89     return result
  90
  91
  92 def _order_json_keys_array(sub_value: JsonTypes, sub_schema: JsonTypes, schema: JsonTypes, path: List) -> List:
  93     if not isinstance(sub_value, list):
  94         raise ValidationError(f'Type of {"".join(_fmt_path(path))} needs to be array (Python list).')
  95     s = sub_schema.get('items', True)
  96     return [_order_json_keys(v, s, schema, path + [i]) for i, v in enumerate(sub_value)]
  97
  98
  99 def _order_json_keys_boolean(sub_value: JsonTypes, sub_schema: JsonTypes, schema: JsonTypes, path: List) -> bool:
 100     if not isinstance(sub_value, bool):
 101         raise ValidationError(f'Type of {_fmt_path(path)} needs to be boolean (Python bool).')
 102     return sub_value
 103
 104
 105 def _order_json_keys_null(sub_value: JsonTypes, sub_schema: JsonTypes, schema: JsonTypes, path: List) -> None:
 106     if sub_value is not None:
 107         raise ValidationError(f'Type of {_fmt_path(path)} needs to be null (Python None).')
 108     return sub_value
 109
 110
 111 def _order_json_keys(sub_value: JsonTypes, sub_schema: JsonTypes, schema: JsonTypes, path: List) -> JsonTypes:
 112     if isinstance(sub_schema, bool):
 113         if sub_schema:
 114             return sub_value
 115         raise ValidationError(f'Value {sub_value} not allowed in {_fmt_path(path)}.')
 116     if isinstance(sub_schema, dict):
 117         sub_schema = _resolve_ref(sub_schema, schema)
 118     return {
 119         'string': _order_json_keys_string,
 120         'number': _order_json_keys_number,
 121         'object': _order_json_keys_object,
 122         'array': _order_json_keys_array,
 123         'boolean': _order_json_keys_boolean,
 124         'null': _order_json_keys_null,
 125     }[sub_schema['type']](sub_value, sub_schema, schema, path)
 126
 127
 128 def order_json_keys(value: JsonTypes, schema: JsonTypes) -> JsonTypes:
 129     return _order_json_keys(value, schema, schema, [])