1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
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: # Fallback if cls cannot be imported
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
|
