codekingpro/portable-devtools

1import json2from collections.abc import Callable, Mapping, Sequence3from contextlib import suppress4from pathlib import PurePath5from typing import Any, ClassVar6 7from .registry import _import_class, get_filesystem_class8from .spec import AbstractFileSystem9 10 11class FilesystemJSONEncoder(json.JSONEncoder):12    include_password: ClassVar[bool] = True13 14    def default(self, o: Any) -> Any:15        if isinstance(o, AbstractFileSystem):16            return o.to_dict(include_password=self.include_password)17        if isinstance(o, PurePath):18            cls = type(o)19            return {"cls": f"{cls.__module__}.{cls.__name__}", "str": str(o)}20 21        return super().default(o)22 23    def make_serializable(self, obj: Any) -> Any:24        """25        Recursively converts an object so that it can be JSON serialized via26        :func:`json.dumps` and :func:`json.dump`, without actually calling27        said functions.28        """29        if isinstance(obj, (str, int, float, bool)):30            return obj31        if isinstance(obj, Mapping):32            return {k: self.make_serializable(v) for k, v in obj.items()}33        if isinstance(obj, Sequence):34            return [self.make_serializable(v) for v in obj]35 36        return self.default(obj)37 38 39class FilesystemJSONDecoder(json.JSONDecoder):40    def __init__(41        self,42        *,43        object_hook: Callable[[dict[str, Any]], Any] | None = None,44        parse_float: Callable[[str], Any] | None = None,45        parse_int: Callable[[str], Any] | None = None,46        parse_constant: Callable[[str], Any] | None = None,47        strict: bool = True,48        object_pairs_hook: Callable[[list[tuple[str, Any]]], Any] | None = None,49    ) -> None:50        self.original_object_hook = object_hook51 52        super().__init__(53            object_hook=self.custom_object_hook,54            parse_float=parse_float,55            parse_int=parse_int,56            parse_constant=parse_constant,57            strict=strict,58            object_pairs_hook=object_pairs_hook,59        )60 61    @classmethod62    def try_resolve_path_cls(cls, dct: dict[str, Any]):63        with suppress(Exception):64            fqp = dct["cls"]65 66            path_cls = _import_class(fqp)67 68            if issubclass(path_cls, PurePath):69                return path_cls70 71        return None72 73    @classmethod74    def try_resolve_fs_cls(cls, dct: dict[str, Any]):75        with suppress(Exception):76            if "cls" in dct:77                try:78                    fs_cls = _import_class(dct["cls"])79                    if issubclass(fs_cls, AbstractFileSystem):80                        return fs_cls81                except Exception:82                    if "protocol" in dct:  # Fallback if cls cannot be imported83                        return get_filesystem_class(dct["protocol"])84 85                    raise86 87        return None88 89    def custom_object_hook(self, dct: dict[str, Any]):90        if "cls" in dct:91            if (obj_cls := self.try_resolve_fs_cls(dct)) is not None:92                return AbstractFileSystem.from_dict(dct)93            if (obj_cls := self.try_resolve_path_cls(dct)) is not None:94                return obj_cls(dct["str"])95 96        if self.original_object_hook is not None:97            return self.original_object_hook(dct)98 99        return dct100 101    def unmake_serializable(self, obj: Any) -> Any:102        """103        Inverse function of :meth:`FilesystemJSONEncoder.make_serializable`.104        """105        if isinstance(obj, dict):106            obj = self.custom_object_hook(obj)107        if isinstance(obj, dict):108            return {k: self.unmake_serializable(v) for k, v in obj.items()}109        if isinstance(obj, (list, tuple)):110            return [self.unmake_serializable(v) for v in obj]111 112        return obj113