|
2 | 2 |
|
3 | 3 | from __future__ import annotations
|
4 | 4 |
|
| 5 | +import json |
| 6 | +import os |
5 | 7 | import re
|
6 | 8 | import typing
|
7 | 9 | from collections import OrderedDict, UserList, defaultdict
|
|
45 | 47 | from frozenlist import FrozenList
|
46 | 48 | from pprintpp import pformat
|
47 | 49 |
|
48 |
| -from pycardano.exception import DeserializeException, SerializeException |
| 50 | +from pycardano.exception import ( |
| 51 | + DeserializeException, |
| 52 | + InvalidKeyTypeException, |
| 53 | + SerializeException, |
| 54 | +) |
49 | 55 | from pycardano.types import check_type, typechecked
|
50 | 56 |
|
51 | 57 | __all__ = [
|
|
63 | 69 | "OrderedSet",
|
64 | 70 | "NonEmptyOrderedSet",
|
65 | 71 | "CodedSerializable",
|
| 72 | + "TextEnvelope", |
66 | 73 | ]
|
67 | 74 |
|
68 | 75 | T = TypeVar("T")
|
@@ -1142,3 +1149,98 @@ def from_primitive(
|
1142 | 1149 | raise DeserializeException(f"Invalid {cls.__name__} type {values[0]}")
|
1143 | 1150 | # Cast using Type[CodedSerializable] instead of cls directly
|
1144 | 1151 | return cast(Type[CodedSerializable], super()).from_primitive(values[1:])
|
| 1152 | + |
| 1153 | + |
| 1154 | +@dataclass(repr=False) |
| 1155 | +class TextEnvelope(CBORSerializable): |
| 1156 | + """A base class for TextEnvelope types that can be saved and loaded as JSON.""" |
| 1157 | + |
| 1158 | + KEY_TYPE = "" |
| 1159 | + DESCRIPTION = "" |
| 1160 | + |
| 1161 | + def __init__( |
| 1162 | + self, |
| 1163 | + payload: Optional[bytes] = None, |
| 1164 | + key_type: Optional[str] = None, |
| 1165 | + description: Optional[str] = None, |
| 1166 | + ): |
| 1167 | + self._payload = payload |
| 1168 | + self._key_type = key_type or self.KEY_TYPE |
| 1169 | + self._description = description or self.DESCRIPTION |
| 1170 | + |
| 1171 | + @property |
| 1172 | + def payload(self) -> bytes: |
| 1173 | + if self._payload is None: |
| 1174 | + self._payload = self.to_cbor() |
| 1175 | + return self._payload |
| 1176 | + |
| 1177 | + @property |
| 1178 | + def key_type(self) -> str: |
| 1179 | + return self._key_type |
| 1180 | + |
| 1181 | + @property |
| 1182 | + def description(self) -> str: |
| 1183 | + return self._description |
| 1184 | + |
| 1185 | + def to_json(self) -> str: |
| 1186 | + """Serialize the key to JSON. |
| 1187 | +
|
| 1188 | + The json output has three fields: "type", "description", and "cborHex". |
| 1189 | +
|
| 1190 | + Returns: |
| 1191 | + str: JSON representation of the key. |
| 1192 | + """ |
| 1193 | + return json.dumps( |
| 1194 | + { |
| 1195 | + "type": self.key_type, |
| 1196 | + "description": self.description, |
| 1197 | + "cborHex": self.to_cbor_hex(), |
| 1198 | + } |
| 1199 | + ) |
| 1200 | + |
| 1201 | + @classmethod |
| 1202 | + def from_json( |
| 1203 | + cls: Type[TextEnvelope], data: str, validate_type=False |
| 1204 | + ) -> TextEnvelope: |
| 1205 | + """Restore a TextEnvelope from a JSON string. |
| 1206 | +
|
| 1207 | + Args: |
| 1208 | + data (str): JSON string. |
| 1209 | + validate_type (bool): Checks whether the type specified in json object is the same |
| 1210 | + as the class's default type. |
| 1211 | +
|
| 1212 | + Returns: |
| 1213 | + Key: The key restored from JSON. |
| 1214 | +
|
| 1215 | + Raises: |
| 1216 | + InvalidKeyTypeException: When `validate_type=True` and the type in json is not equal to the default type |
| 1217 | + of the Key class used. |
| 1218 | + """ |
| 1219 | + obj = json.loads(data) |
| 1220 | + |
| 1221 | + if validate_type and obj["type"] != cls.KEY_TYPE: |
| 1222 | + raise InvalidKeyTypeException( |
| 1223 | + f"Expect key type: {cls.KEY_TYPE}, got {obj['type']} instead." |
| 1224 | + ) |
| 1225 | + |
| 1226 | + k = cls.from_cbor(obj["cborHex"]) |
| 1227 | + |
| 1228 | + assert isinstance(k, cls) |
| 1229 | + |
| 1230 | + k._key_type = obj["type"] |
| 1231 | + k._description = obj["description"] |
| 1232 | + k._payload = k.to_cbor() |
| 1233 | + |
| 1234 | + return k |
| 1235 | + |
| 1236 | + def save(self, path: str): |
| 1237 | + if os.path.isfile(path): |
| 1238 | + if os.stat(path).st_size > 0: |
| 1239 | + raise IOError(f"File {path} already exists!") |
| 1240 | + with open(path, "w") as f: |
| 1241 | + f.write(self.to_json()) |
| 1242 | + |
| 1243 | + @classmethod |
| 1244 | + def load(cls, path: str): |
| 1245 | + with open(path) as f: |
| 1246 | + return cls.from_json(f.read()) |
0 commit comments