repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
public Clawd ADK gateway launch mirror
stars
latest
clone command
git clone gitlawb://did:key:z6Mkq5mY...iFZ5/my-project-publ...git clone gitlawb://did:key:z6Mkq5mY.../my-project-publ...2fa351d6docs: add automaton and perps launch sources16d ago| #1 | import json |
| #2 | import logging |
| #3 | from string import Template |
| #4 | from typing import Any, Type, TypeVar, Union |
| #5 | |
| #6 | T = TypeVar("T", bound="JSONSerializable") |
| #7 | |
| #8 | # NOTE: Through inheritance, all of our classes should be children of JSONSerializable. (highest level) |
| #9 | # NOTE: The @register_deserializable decorator should be added to all user facing child classes. (lowest level) |
| #10 | |
| #11 | logger = logging.getLogger(__name__) |
| #12 | |
| #13 | |
| #14 | def register_deserializable(cls: Type[T]) -> Type[T]: |
| #15 | """ |
| #16 | A class decorator to register a class as deserializable. |
| #17 | |
| #18 | When a class is decorated with @register_deserializable, it becomes |
| #19 | a part of the set of classes that the JSONSerializable class can |
| #20 | deserialize. |
| #21 | |
| #22 | Deserialization is in essence loading attributes from a json file. |
| #23 | This decorator is a security measure put in place to make sure that |
| #24 | you don't load attributes that were initially part of another class. |
| #25 | |
| #26 | Example: |
| #27 | @register_deserializable |
| #28 | class ChildClass(JSONSerializable): |
| #29 | def __init__(self, ...): |
| #30 | # initialization logic |
| #31 | |
| #32 | Args: |
| #33 | cls (Type): The class to be registered. |
| #34 | |
| #35 | Returns: |
| #36 | Type: The same class, after registration. |
| #37 | """ |
| #38 | JSONSerializable._register_class_as_deserializable(cls) |
| #39 | return cls |
| #40 | |
| #41 | |
| #42 | class JSONSerializable: |
| #43 | """ |
| #44 | A class to represent a JSON serializable object. |
| #45 | |
| #46 | This class provides methods to serialize and deserialize objects, |
| #47 | as well as to save serialized objects to a file and load them back. |
| #48 | """ |
| #49 | |
| #50 | _deserializable_classes = set() # Contains classes that are whitelisted for deserialization. |
| #51 | |
| #52 | def serialize(self) -> str: |
| #53 | """ |
| #54 | Serialize the object to a JSON-formatted string. |
| #55 | |
| #56 | Returns: |
| #57 | str: A JSON string representation of the object. |
| #58 | """ |
| #59 | try: |
| #60 | return json.dumps(self, default=self._auto_encoder, ensure_ascii=False) |
| #61 | except Exception as e: |
| #62 | logger.error(f"Serialization error: {e}") |
| #63 | return "{}" |
| #64 | |
| #65 | @classmethod |
| #66 | def deserialize(cls, json_str: str) -> Any: |
| #67 | """ |
| #68 | Deserialize a JSON-formatted string to an object. |
| #69 | If it fails, a default class is returned instead. |
| #70 | Note: This *returns* an instance, it's not automatically loaded on the calling class. |
| #71 | |
| #72 | Example: |
| #73 | app = App.deserialize(json_str) |
| #74 | |
| #75 | Args: |
| #76 | json_str (str): A JSON string representation of an object. |
| #77 | |
| #78 | Returns: |
| #79 | Object: The deserialized object. |
| #80 | """ |
| #81 | try: |
| #82 | return json.loads(json_str, object_hook=cls._auto_decoder) |
| #83 | except Exception as e: |
| #84 | logger.error(f"Deserialization error: {e}") |
| #85 | # Return a default instance in case of failure |
| #86 | return cls() |
| #87 | |
| #88 | @staticmethod |
| #89 | def _auto_encoder(obj: Any) -> Union[dict[str, Any], None]: |
| #90 | """ |
| #91 | Automatically encode an object for JSON serialization. |
| #92 | |
| #93 | Args: |
| #94 | obj (Object): The object to be encoded. |
| #95 | |
| #96 | Returns: |
| #97 | dict: A dictionary representation of the object. |
| #98 | """ |
| #99 | if hasattr(obj, "__dict__"): |
| #100 | dct = {} |
| #101 | for key, value in obj.__dict__.items(): |
| #102 | try: |
| #103 | # Recursive: If the value is an instance of a subclass of JSONSerializable, |
| #104 | # serialize it using the JSONSerializable serialize method. |
| #105 | if isinstance(value, JSONSerializable): |
| #106 | serialized_value = value.serialize() |
| #107 | # The value is stored as a serialized string. |
| #108 | dct[key] = json.loads(serialized_value) |
| #109 | # Custom rules (subclass is not json serializable by default) |
| #110 | elif isinstance(value, Template): |
| #111 | dct[key] = {"__type__": "Template", "data": value.template} |
| #112 | # Future custom types we can follow a similar pattern |
| #113 | # elif isinstance(value, SomeOtherType): |
| #114 | # dct[key] = { |
| #115 | # "__type__": "SomeOtherType", |
| #116 | # "data": value.some_method() |
| #117 | # } |
| #118 | # NOTE: Keep in mind that this logic needs to be applied to the decoder too. |
| #119 | else: |
| #120 | json.dumps(value) # Try to serialize the value. |
| #121 | dct[key] = value |
| #122 | except TypeError: |
| #123 | pass # If it fails, simply pass to skip this key-value pair of the dictionary. |
| #124 | |
| #125 | dct["__class__"] = obj.__class__.__name__ |
| #126 | return dct |
| #127 | raise TypeError(f"Object of type {type(obj)} is not JSON serializable") |
| #128 | |
| #129 | @classmethod |
| #130 | def _auto_decoder(cls, dct: dict[str, Any]) -> Any: |
| #131 | """ |
| #132 | Automatically decode a dictionary to an object during JSON deserialization. |
| #133 | |
| #134 | Args: |
| #135 | dct (dict): The dictionary representation of an object. |
| #136 | |
| #137 | Returns: |
| #138 | Object: The decoded object or the original dictionary if decoding is not possible. |
| #139 | """ |
| #140 | class_name = dct.pop("__class__", None) |
| #141 | if class_name: |
| #142 | if not hasattr(cls, "_deserializable_classes"): # Additional safety check |
| #143 | raise AttributeError(f"`{class_name}` has no registry of allowed deserializations.") |
| #144 | if class_name not in {cl.__name__ for cl in cls._deserializable_classes}: |
| #145 | raise KeyError(f"Deserialization of class `{class_name}` is not allowed.") |
| #146 | target_class = next((cl for cl in cls._deserializable_classes if cl.__name__ == class_name), None) |
| #147 | if target_class: |
| #148 | obj = target_class.__new__(target_class) |
| #149 | for key, value in dct.items(): |
| #150 | if isinstance(value, dict) and "__type__" in value: |
| #151 | if value["__type__"] == "Template": |
| #152 | value = Template(value["data"]) |
| #153 | # For future custom types we can follow a similar pattern |
| #154 | # elif value["__type__"] == "SomeOtherType": |
| #155 | # value = SomeOtherType.some_constructor(value["data"]) |
| #156 | default_value = getattr(target_class, key, None) |
| #157 | setattr(obj, key, value or default_value) |
| #158 | return obj |
| #159 | return dct |
| #160 | |
| #161 | def save_to_file(self, filename: str) -> None: |
| #162 | """ |
| #163 | Save the serialized object to a file. |
| #164 | |
| #165 | Args: |
| #166 | filename (str): The path to the file where the object should be saved. |
| #167 | """ |
| #168 | with open(filename, "w", encoding="utf-8") as f: |
| #169 | f.write(self.serialize()) |
| #170 | |
| #171 | @classmethod |
| #172 | def load_from_file(cls, filename: str) -> Any: |
| #173 | """ |
| #174 | Load and deserialize an object from a file. |
| #175 | |
| #176 | Args: |
| #177 | filename (str): The path to the file from which the object should be loaded. |
| #178 | |
| #179 | Returns: |
| #180 | Object: The deserialized object. |
| #181 | """ |
| #182 | with open(filename, "r", encoding="utf-8") as f: |
| #183 | json_str = f.read() |
| #184 | return cls.deserialize(json_str) |
| #185 | |
| #186 | @classmethod |
| #187 | def _register_class_as_deserializable(cls, target_class: Type[T]) -> None: |
| #188 | """ |
| #189 | Register a class as deserializable. This is a classmethod and globally shared. |
| #190 | |
| #191 | This method adds the target class to the set of classes that |
| #192 | can be deserialized. This is a security measure to ensure only |
| #193 | whitelisted classes are deserialized. |
| #194 | |
| #195 | Args: |
| #196 | target_class (Type): The class to be registered. |
| #197 | """ |
| #198 | cls._deserializable_classes.add(target_class) |
| #199 |