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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
from __future__ import annotations
import copy
import json
from datetime import date
from functools import wraps
from pathlib import Path
from typing import Any, Callable, TypeVar
from sec_certs import constants
T = TypeVar("T", bound="ComplexSerializableType")
class SerializationError(Exception):
pass
class ComplexSerializableType:
__slots__: tuple[str]
def __init__(self, *args, **kwargs):
pass
# Ideally, the serialized_fields would be an class variable referencing itself, but that it virtually impossible
# to achieve without using metaclasses. Not to complicate the code, we choose instance variable.
@property
def serialized_attributes(self) -> list[str]:
if hasattr(self, "__slots__") and self.__slots__:
return list(self.__slots__)
return list(self.__dict__.keys())
def to_dict(self) -> dict[str, Any]:
if hasattr(self, "__slots__") and self.__slots__:
return {
key: copy.deepcopy(getattr(self, key)) for key in self.__slots__ if key in self.serialized_attributes
}
return {key: val for key, val in copy.deepcopy(self.__dict__).items() if key in self.serialized_attributes}
@classmethod
def from_dict(cls: type[T], dct: dict) -> T:
try:
return cls(**dct)
except TypeError as e:
raise TypeError(f"Dict: {dct} on {cls.__mro__}") from e
def to_json(self, output_path: str | Path | None = None) -> None:
if not output_path and (not hasattr(self, "json_path") or not self.json_path): # type: ignore
raise SerializationError(
f"The object {self} of type {self.__class__} does not have json_path attribute set but to_json() was called without an argument."
)
if not output_path:
output_path = self.json_path # type: ignore
if self.json_path == constants.DUMMY_NONEXISTING_PATH: # type: ignore
raise SerializationError(f"json_path attribute for '{get_class_fullname(self)}' was not yet set.")
if hasattr(self, "root_dir") and self.root_dir == constants.DUMMY_NONEXISTING_PATH: # type: ignore
raise SerializationError(f"root_dir attribute for '{get_class_fullname(self)}' was not yet set.")
if Path(output_path).is_dir(): # type: ignore
raise SerializationError("output path for json cannot be directory.")
# false positive MyPy warning, cannot be None
with Path(output_path).open("w") as handle: # type: ignore
json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False)
@classmethod
def from_json(cls: type[T], input_path: str | Path) -> T:
input_path = Path(input_path)
with input_path.open("r") as handle:
obj = json.load(handle, cls=CustomJSONDecoder)
return obj
# Decorator for serialization
def serialize(func: Callable):
@wraps(func)
def inner_func(*args, **kwargs):
if not args or not issubclass(type(args[0]), ComplexSerializableType):
raise ValueError(
"@serialize decorator is to be used only on instance methods of ComplexSerializableType child classes."
)
if hasattr(args[0], "_root_dir") and args[0]._root_dir == constants.DUMMY_NONEXISTING_PATH:
raise SerializationError(
"The invoked method requires dataset serialization. Cannot serialize without root_dir set. You can set it with obj.root_dir = ..."
)
update_json = kwargs.pop("update_json", True)
result = func(*args, **kwargs)
if update_json:
args[0].to_json()
return result
return inner_func
def get_class_fullname(obj: Any) -> str:
if isinstance(obj, type):
klass = obj
else:
klass = obj.__class__
module = klass.__module__
if module == "builtins":
return klass.__qualname__
return module + "." + klass.__qualname__
class CustomJSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, ComplexSerializableType):
return {**{"_type": get_class_fullname(obj)}, **obj.to_dict()}
if isinstance(obj, dict):
return obj
if isinstance(obj, set):
return {"_type": "Set", "elements": sorted(list(obj))}
if isinstance(obj, frozenset):
return sorted(list(obj))
if isinstance(obj, date):
return str(obj)
if isinstance(obj, Path):
return str(obj)
return super().default(obj)
class CustomJSONDecoder(json.JSONDecoder):
"""
Custom JSONDecoder. Any complex object that should be de-serializable must inherit directly from class
ComplexSerializableType (nested inheritance does not currently work (because x.__subclassess__() prints only direct
subclasses. Any such class must implement methods to_dict() and from_dict(). These are used to drive serialization.
"""
def __init__(self, *args, **kwargs):
json.JSONDecoder.__init__(self, object_hook=self.object_hook, *args, **kwargs)
self.serializable_complex_types = {get_class_fullname(x): x for x in ComplexSerializableType.__subclasses__()}
def object_hook(self, obj):
if "_type" in obj and obj["_type"] == "Set":
return set(obj["elements"])
if "_type" in obj and obj["_type"] in self.serializable_complex_types.keys():
complex_type = obj.pop("_type")
return self.serializable_complex_types[complex_type].from_dict(obj)
elif "_type" in obj:
raise SerializationError(f"JSONDecoder doesn't know how to handle {obj}")
return obj
|
