diff options
Diffstat (limited to 'src/hydrilla/json_instances.py')
-rw-r--r-- | src/hydrilla/json_instances.py | 34 |
1 files changed, 21 insertions, 13 deletions
diff --git a/src/hydrilla/json_instances.py b/src/hydrilla/json_instances.py index be8dbc6..8bec808 100644 --- a/src/hydrilla/json_instances.py +++ b/src/hydrilla/json_instances.py @@ -127,11 +127,14 @@ schema_paths.update([(f'https://hydrilla.koszko.org/schemas/{name}', path) schemas: dict[Path, dict[str, t.Any]] = {} +class UnknownSchemaError(HaketiloException): + pass + def _get_schema(schema_name: str) -> dict[str, t.Any]: """Return loaded JSON of the requested schema. Cache results.""" path = schema_paths.get(schema_name) if path is None: - raise HaketiloException(_('unknown_schema_{}').format(schema_name)) + raise UnknownSchemaError(_('unknown_schema_{}').format(schema_name)) if path not in schemas: schemas[path] = json.loads(path.read_text()) @@ -159,28 +162,33 @@ def parse_instance(text: str) -> object: """Parse 'text' as JSON with additional '//' comments support.""" return json.loads(strip_json_comments(text)) -InstanceOrPathOrIO = t.Union[Path, str, io.TextIOBase, dict[str, t.Any]] +InstanceSource = t.Union[Path, str, io.TextIOBase, dict[str, t.Any], bytes] -def read_instance(instance_or_path: InstanceOrPathOrIO) -> object: +def read_instance(instance_or_path: InstanceSource) -> object: """....""" if isinstance(instance_or_path, dict): return instance_or_path - if isinstance(instance_or_path, io.TextIOBase): - handle = instance_or_path + if isinstance(instance_or_path, bytes): + encoding = json.detect_encoding(instance_or_path) + text = instance_or_path.decode(encoding) + elif isinstance(instance_or_path, io.TextIOBase): + try: + text = instance_or_path.read() + finally: + instance_or_path.close() else: - handle = t.cast(io.TextIOBase, open(instance_or_path, 'rt')) - - try: - text = handle.read() - finally: - handle.close() + text = Path(instance_or_path).read_text() try: return parse_instance(text) except: - fmt = _('err.util.text_in_{}_not_valid_json') - raise HaketiloException(fmt.format(instance_or_path)) + if isinstance(instance_or_path, str) or \ + isinstance(instance_or_path, Path): + fmt = _('err.util.text_in_{}_not_valid_json') + raise HaketiloException(fmt.format(instance_or_path)) + else: + raise HaketiloException(_('err.util.text_not_valid_json')) def get_schema_version(instance: object) -> tuple[int, ...]: """ |