|
import json |
|
from contextlib import suppress |
|
from pathlib import PurePath |
|
from typing import ( |
|
Any, |
|
Callable, |
|
ClassVar, |
|
Dict, |
|
List, |
|
Mapping, |
|
Optional, |
|
Sequence, |
|
Tuple, |
|
) |
|
|
|
from .registry import _import_class, get_filesystem_class |
|
from .spec import AbstractFileSystem |
|
|
|
|
|
class FilesystemJSONEncoder(json.JSONEncoder): |
|
include_password: ClassVar[bool] = True |
|
|
|
def default(self, o: Any) -> Any: |
|
if isinstance(o, AbstractFileSystem): |
|
return o.to_dict(include_password=self.include_password) |
|
if isinstance(o, PurePath): |
|
cls = type(o) |
|
return {"cls": f"{cls.__module__}.{cls.__name__}", "str": str(o)} |
|
|
|
return super().default(o) |
|
|
|
def make_serializable(self, obj: Any) -> Any: |
|
""" |
|
Recursively converts an object so that it can be JSON serialized via |
|
:func:`json.dumps` and :func:`json.dump`, without actually calling |
|
said functions. |
|
""" |
|
if isinstance(obj, (str, int, float, bool)): |
|
return obj |
|
if isinstance(obj, Mapping): |
|
return {k: self.make_serializable(v) for k, v in obj.items()} |
|
if isinstance(obj, Sequence): |
|
return [self.make_serializable(v) for v in obj] |
|
|
|
return self.default(obj) |
|
|
|
|
|
class FilesystemJSONDecoder(json.JSONDecoder): |
|
def __init__( |
|
self, |
|
*, |
|
object_hook: Optional[Callable[[Dict[str, Any]], Any]] = None, |
|
parse_float: Optional[Callable[[str], Any]] = None, |
|
parse_int: Optional[Callable[[str], Any]] = None, |
|
parse_constant: Optional[Callable[[str], Any]] = None, |
|
strict: bool = True, |
|
object_pairs_hook: Optional[Callable[[List[Tuple[str, Any]]], Any]] = None, |
|
) -> None: |
|
self.original_object_hook = object_hook |
|
|
|
super().__init__( |
|
object_hook=self.custom_object_hook, |
|
parse_float=parse_float, |
|
parse_int=parse_int, |
|
parse_constant=parse_constant, |
|
strict=strict, |
|
object_pairs_hook=object_pairs_hook, |
|
) |
|
|
|
@classmethod |
|
def try_resolve_path_cls(cls, dct: Dict[str, Any]): |
|
with suppress(Exception): |
|
fqp = dct["cls"] |
|
|
|
path_cls = _import_class(fqp) |
|
|
|
if issubclass(path_cls, PurePath): |
|
return path_cls |
|
|
|
return None |
|
|
|
@classmethod |
|
def try_resolve_fs_cls(cls, dct: Dict[str, Any]): |
|
with suppress(Exception): |
|
if "cls" in dct: |
|
try: |
|
fs_cls = _import_class(dct["cls"]) |
|
if issubclass(fs_cls, AbstractFileSystem): |
|
return fs_cls |
|
except Exception: |
|
if "protocol" in dct: |
|
return get_filesystem_class(dct["protocol"]) |
|
|
|
raise |
|
|
|
return None |
|
|
|
def custom_object_hook(self, dct: Dict[str, Any]): |
|
if "cls" in dct: |
|
if (obj_cls := self.try_resolve_fs_cls(dct)) is not None: |
|
return AbstractFileSystem.from_dict(dct) |
|
if (obj_cls := self.try_resolve_path_cls(dct)) is not None: |
|
return obj_cls(dct["str"]) |
|
|
|
if self.original_object_hook is not None: |
|
return self.original_object_hook(dct) |
|
|
|
return dct |
|
|
|
def unmake_serializable(self, obj: Any) -> Any: |
|
""" |
|
Inverse function of :meth:`FilesystemJSONEncoder.make_serializable`. |
|
""" |
|
if isinstance(obj, dict): |
|
obj = self.custom_object_hook(obj) |
|
if isinstance(obj, dict): |
|
return {k: self.unmake_serializable(v) for k, v in obj.items()} |
|
if isinstance(obj, (list, tuple)): |
|
return [self.unmake_serializable(v) for v in obj] |
|
|
|
return obj |
|
|
