edq.util.json

  1"""
  2This file standardizes how we write and read JSON.
  3Specifically, we try to be flexible when reading (using JSON5),
  4and strict when writing (using vanilla JSON).
  5"""
  6
  7import enum
  8import gzip
  9import io
 10import json
 11import os
 12import typing
 13
 14import json5
 15
 16import edq.util.dirent
 17
 18class DictConverter():
 19    """
 20    A base class for class that can represent (serialize) and reconstruct (deserialize) themselves as/from a dict.
 21    The intention is that the dict can then be cleanly converted to/from JSON.
 22
 23    General (but inefficient) implementations of several core Python equality, comparison, and representation methods are provided.
 24    """
 25
 26    def to_dict(self) -> typing.Dict[str, typing.Any]:
 27        """
 28        Return a dict that can be used to represent this object.
 29        If the dict is passed to from_dict(), an identical object should be reconstructed.
 30
 31        A general (but inefficient) implementation is provided by default.
 32        """
 33
 34        return vars(self).copy()
 35
 36    @classmethod
 37    # Note that `typing.Self` is returned, but that is introduced in Python 3.12.
 38    def from_dict(cls, data: typing.Dict[str, typing.Any]) -> typing.Any:
 39        """
 40        Return an instance of this subclass created using the given dict.
 41        If the dict came from to_dict(), the returned object should be identical to the original.
 42
 43        A general (but inefficient) implementation is provided by default.
 44        """
 45
 46        return cls(**data)
 47
 48    def __eq__(self, other: object) -> bool:
 49        """
 50        Check for equality.
 51
 52        This check uses to_dict() and compares the results.
 53        This may not be complete or efficient depending on the child class.
 54        """
 55
 56        # Note the hard type check (done so we can keep this method general).
 57        if (type(self) != type(other)):  # pylint: disable=unidiomatic-typecheck
 58            return False
 59
 60        return bool(self.to_dict() == other.to_dict())  # type: ignore[attr-defined]
 61
 62    def __lt__(self, other: 'DictConverter') -> bool:
 63        return dumps(self) < dumps(other)
 64
 65    def __hash__(self) -> int:
 66        return hash(dumps(self))
 67
 68    def __str__(self) -> str:
 69        return dumps(self)
 70
 71    def __repr__(self) -> str:
 72        return dumps(self)
 73
 74def _custom_handle(value: typing.Any) -> typing.Union[typing.Dict[str, typing.Any], str]:
 75    """
 76    Handle objects that are not JSON serializable by default,
 77    e.g., calling vars() on an object.
 78    """
 79
 80    if (isinstance(value, DictConverter)):
 81        return value.to_dict()
 82
 83    if (isinstance(value, enum.Enum)):
 84        return str(value)
 85
 86    if (hasattr(value, '__dict__')):
 87        return dict(vars(value))
 88
 89    raise ValueError(f"Could not JSON serialize object: '{value}'.")
 90
 91def load(
 92        file_obj: typing.TextIO,
 93        strict: bool = False,
 94        gzipped: bool = False,
 95        encoding: str = edq.util.dirent.DEFAULT_ENCODING,
 96        **kwargs: typing.Any) -> typing.Any:
 97    """
 98    Load a file object/handler as JSON.
 99    If strict is set, then use standard Python JSON,
100    otherwise use JSON5.
101
102    If `gzipped` is set, the file object is treated as a gzipped bytes stream (e.g. `open('test.json.gz', 'rb')`).
103    """
104
105    if (gzipped):
106        binary_file_obj = gzip.GzipFile(fileobj = file_obj)  # type: ignore[call-overload]
107        file_obj = io.TextIOWrapper(binary_file_obj, encoding = encoding)
108
109    if (strict):
110        return json.load(file_obj, **kwargs)
111
112    return json5.load(file_obj, **kwargs)
113
114def loads(text: str, strict: bool = False, **kwargs: typing.Any) -> typing.Any:
115    """
116    Load a string as JSON.
117    If strict is set, then use standard Python JSON,
118    otherwise use JSON5.
119    """
120
121    if (strict):
122        return json.loads(text, **kwargs)
123
124    return json5.loads(text, **kwargs)
125
126def load_path(
127        path: str,
128        strict: bool = False,
129        gzipped: typing.Union[bool, None] = None,
130        encoding: str = edq.util.dirent.DEFAULT_ENCODING,
131        **kwargs: typing.Any) -> typing.Any:
132    """
133    Load a file path as JSON.
134    If strict is set, then use standard Python JSON,
135    otherwise use JSON5.
136
137    If `gzipped` is not set, the behavior is guessed from the extension (".gz").
138    """
139
140    if (not os.path.exists(path)):
141        raise FileNotFoundError(f"File does not exist: '{path}'.")
142
143    if (os.path.isdir(path)):
144        raise IsADirectoryError(f"Cannot open JSON file, expected a file but got a directory at '{path}'.")
145
146    if (gzipped is None):
147        gzipped = (os.path.splitext(path)[-1] == '.gz')
148
149    open_func = open
150    if (gzipped):
151        open_func = gzip.open  # type: ignore[assignment]
152
153    with open_func(path, 'rt', encoding = encoding) as file:
154        try:
155            return load(file, strict = strict, **kwargs)
156        except Exception as ex:
157            raise ValueError(f"Failed to read JSON file '{path}'.") from ex
158
159def loads_object(text: str, cls: typing.Type[DictConverter], **kwargs: typing.Any) -> DictConverter:
160    """ Load a JSON string into an object (which is a subclass of DictConverter). """
161
162    data = loads(text, **kwargs)
163    if (not isinstance(data, dict)):
164        raise ValueError(f"JSON to load into an object is not a dict, found '{type(data)}'.")
165
166    return cls.from_dict(data)  # type: ignore[no-any-return]
167
168def load_object_path(path: str, cls: typing.Type[DictConverter], **kwargs: typing.Any) -> DictConverter:
169    """ Load a JSON file into an object (which is a subclass of DictConverter). """
170
171    data = load_path(path, **kwargs)
172    if (not isinstance(data, dict)):
173        raise ValueError(f"JSON to load into an object is not a dict, found '{type(data)}'.")
174
175    return cls.from_dict(data)  # type: ignore[no-any-return]
176
177def dump(
178        data: typing.Any,
179        file_obj: typing.TextIO,
180        default: typing.Union[typing.Callable, None] = _custom_handle,
181        sort_keys: bool = True,
182        **kwargs: typing.Any) -> None:
183    """ Dump an object as a JSON file object. """
184
185    json.dump(data, file_obj, default = default, sort_keys = sort_keys, **kwargs)
186
187def dumps(
188        data: typing.Any,
189        default: typing.Union[typing.Callable, None] = _custom_handle,
190        sort_keys: bool = True,
191        **kwargs: typing.Any) -> str:
192    """ Dump an object as a JSON string. """
193
194    return json.dumps(data, default = default, sort_keys = sort_keys, **kwargs)
195
196def dump_path(
197        data: typing.Any,
198        path: str,
199        default: typing.Union[typing.Callable, None] = _custom_handle,
200        sort_keys: bool = True,
201        gzipped: typing.Union[bool, None] = None,
202        encoding: str = edq.util.dirent.DEFAULT_ENCODING,
203        **kwargs: typing.Any) -> None:
204    """
205    Dump an object as a JSON file.
206
207    If `gzipped` is not set, the behavior is guessed from the extension (".gz").
208    """
209
210    if (gzipped is None):
211        gzipped = (os.path.splitext(path)[-1] == '.gz')
212
213    open_func = open
214    if (gzipped):
215        open_func = gzip.open  # type: ignore[assignment]
216
217    with open_func(path, 'wt', encoding = encoding) as file:
218        dump(data, file, default = default, sort_keys = sort_keys, **kwargs)  # type: ignore[arg-type]