Add 'to_json' and 'from_json' to PlutusData. This helps us to read and write datum when it can only be communicated in json format by some tools, such cardano-db-sync and cardano-cli.

cffls · cffls · commit bd82b6b4e992 · 2022-03-10T16:35:59.000-08:00
diff --git a/pycardano/plutus.py b/pycardano/plutus.py
@@ -3,9 +3,10 @@
 from __future__ import annotations
 
 import inspect
+import json
 from dataclasses import dataclass, fields
 from enum import Enum
-from typing import ClassVar, Optional, Type, TypeVar
+from typing import ClassVar, Optional, Type, TypeVar, Union
 
 import cbor2
 from cbor2 import CBORTag
@@ -315,6 +316,117 @@ def hash(self) -> DatumHash:
             blake2b(self.to_cbor("bytes"), DATUM_HASH_SIZE, encoder=RawEncoder)
         )
 
+    def to_json(self, **kwargs) -> str:
+        """Convert to a json string
+
+        Args:
+            **kwargs: Extra key word arguments to be passed to `json.dumps()`
+
+        Returns:
+            str: a JSON encoded PlutusData.
+        """
+
+        def _dfs(obj):
+            """
+            Reference of Haskell's implementation:
+            https://github.com/input-output-hk/cardano-node/blob/baa9b5e59c5d448d475f94cc88a31a5857c2bda5/cardano-api/
+            src/Cardano/Api/ScriptData.hs#L449-L474
+            """
+            if isinstance(obj, int):
+                return {"int": obj}
+            elif isinstance(obj, bytes):
+                return {"bytes": obj.hex()}
+            elif isinstance(obj, list):
+                return [_dfs(item) for item in obj]
+            elif isinstance(obj, IndefiniteList):
+                return {"list": [_dfs(item) for item in obj.items]}
+            elif isinstance(obj, dict):
+                return {"map": [{"v": _dfs(v), "k": _dfs(k)} for k, v in obj.items()]}
+            elif isinstance(obj, PlutusData):
+                return {
+                    "constructor": obj.CONSTR_ID,
+                    "fields": _dfs([getattr(obj, f.name) for f in fields(obj)]),
+                }
+            else:
+                raise TypeError(f"Unexpected type {type(obj)}")
+
+        return json.dumps(_dfs(self), **kwargs)
+
+    @classmethod
+    def from_dict(cls: Type[PData], data: dict) -> PData:
+        """Convert a dictionary to PlutusData
+
+        Args:
+            data (dict): A dictionary.
+
+        Returns:
+            PlutusData: Restored PlutusData.
+        """
+
+        def _dfs(obj):
+            if isinstance(obj, dict):
+                if "constructor" in obj:
+                    if obj["constructor"] != cls.CONSTR_ID:
+                        raise DeserializeException(
+                            f"Mismatch between constructors, expect: {cls.CONSTR_ID}, "
+                            f"got: {obj['constructor']} instead."
+                        )
+                    converted_fields = []
+                    for f, f_info in zip(obj["fields"], fields(cls)):
+                        if inspect.isclass(f_info.type) and issubclass(
+                            f_info.type, PlutusData
+                        ):
+                            converted_fields.append(f_info.type.from_dict(f))
+                        elif (
+                            hasattr(f_info.type, "__origin__")
+                            and f_info.type.__origin__ is Union
+                        ):
+                            t_args = f_info.type.__args__
+                            found_match = False
+                            for t in t_args:
+                                if (
+                                    inspect.isclass(t)
+                                    and issubclass(t, PlutusData)
+                                    and t.CONSTR_ID == f["constructor"]
+                                ):
+                                    converted_fields.append(t.from_dict(f))
+                                    found_match = True
+                                    break
+                            if not found_match:
+                                raise DeserializeException(
+                                    f"Unexpected data structure: {f}."
+                                )
+                        else:
+                            converted_fields.append(_dfs(f))
+                    return cls(*converted_fields)
+                elif "map" in obj:
+                    return {_dfs(pair["k"]): _dfs(pair["v"]) for pair in obj["map"]}
+                elif "int" in obj:
+                    return obj["int"]
+                elif "bytes" in obj:
+                    return bytes.fromhex(obj["bytes"])
+                elif "list" in obj:
+                    return IndefiniteList([_dfs(item) for item in obj["list"]])
+                else:
+                    raise DeserializeException(f"Unexpected data structure: {obj}")
+            else:
+                raise TypeError(f"Unexpected data type: {type(obj)}")
+
+        return _dfs(data)
+
+    @classmethod
+    def from_json(cls: Type[PData], data: str) -> PData:
+        """Restore a json encoded string to a PlutusData.
+
+        Args:
+            data (str): An encoded json string.
+
+        Returns:
+            PlutusData: The restored PlutusData.
+        """
+        obj = json.loads(data)
+        return cls.from_dict(obj)
+
 
 class RedeemerTag(CBORSerializable, Enum):
     """
diff --git a/pycardano/serialization.py b/pycardano/serialization.py
@@ -143,7 +143,17 @@ def to_primitive(self) -> Primitive:
                 CBOR primitive types.
         """
         result = self.to_shallow_primitive()
-        container_types = (dict, OrderedDict, defaultdict, set, frozenset, tuple, list)
+        container_types = (
+            dict,
+            OrderedDict,
+            defaultdict,
+            set,
+            frozenset,
+            tuple,
+            list,
+            CBORTag,
+            IndefiniteList,
+        )
 
         def _helper(value):
             if isinstance(value, CBORSerializable):
@@ -169,6 +179,8 @@ def _dfs(value):
                 return tuple([_helper(k) for k in value])
             elif isinstance(value, list):
                 return [_helper(k) for k in value]
+            elif isinstance(value, IndefiniteList):
+                return IndefiniteList([_helper(k) for k in value.items])
             elif isinstance(value, CBORTag):
                 return CBORTag(value.tag, _helper(value.value))
             else:
diff --git a/test/pycardano/test_plutus.py b/test/pycardano/test_plutus.py
@@ -1,7 +1,10 @@
 from dataclasses import dataclass
 from test.pycardano.util import check_two_way_cbor
-from typing import List, Union
+from typing import Union
 
+import pytest
+
+from pycardano.exception import DeserializeException, SerializeException
 from pycardano.plutus import (
     COST_MODELS,
     ExecutionUnits,
@@ -67,6 +70,75 @@ def test_plutus_data():
     check_two_way_cbor(my_vesting)
 
 
+def test_plutus_data_json():
+    key_hash = bytes.fromhex("c2ff616e11299d9094ce0a7eb5b7284b705147a822f4ffbd471f971a")
+    deadline = 1643235300000
+    testa = BigTest(MyTest(123, b"1234", IndefiniteList([4, 5, 6]), {1: b"1", 2: b"2"}))
+    testb = LargestTest()
+
+    my_vesting = VestingParam(
+        beneficiary=key_hash, deadline=deadline, testa=testa, testb=testb
+    )
+
+    encoded_json = my_vesting.to_json(separators=(",", ":"))
+
+    assert (
+        '{"constructor":1,"fields":[{"bytes":"c2ff616e11299d9094ce0a7eb5b7284b705147a822f4ffbd471f971a"},'
+        '{"int":1643235300000},{"constructor":8,"fields":[{"constructor":130,"fields":[{"int":123},'
+        '{"bytes":"31323334"},{"list":[{"int":4},{"int":5},{"int":6}]},{"map":[{"v":{"bytes":"31"},'
+        '"k":{"int":1}},{"v":{"bytes":"32"},"k":{"int":2}}]}]}]},{"constructor":9,"fields":[]}]}'
+        == encoded_json
+    )
+
+    assert my_vesting == VestingParam.from_json(encoded_json)
+
+
+def test_plutus_data_to_json_wrong_type():
+    test = MyTest(123, b"1234", IndefiniteList([4, 5, 6]), {1: b"1", 2: b"2"})
+    test.a = "123"
+    with pytest.raises(TypeError):
+        test.to_json()
+
+
+def test_plutus_data_from_json_wrong_constructor():
+    test = (
+        '{"constructor": 129, "fields": [{"int": 123}, {"bytes": "31323334"}, '
+        '{"list": [{"int": 4}, {"int": 5}, {"int": 6}]}, {"map": [{"v": {"bytes": "31"}, '
+        '"k": {"int": 1}}, {"v": {"bytes": "32"}, "k": {"int": 2}}]}]}'
+    )
+    with pytest.raises(DeserializeException):
+        MyTest.from_json(test)
+
+    test2 = (
+        '{"constructor":1,"fields":[{"bytes":"c2ff616e11299d9094ce0a7eb5b7284b705147a822f4ffbd471f971a"},'
+        '{"int":1643235300000},{"constructor":22,"fields":[{"constructor":130,"fields":[{"int":123},'
+        '{"bytes":"31323334"},{"list":[{"int":4},{"int":5},{"int":6}]},{"map":[{"v":{"bytes":"31"},'
+        '"k":{"int":1}},{"v":{"bytes":"32"},"k":{"int":2}}]}]}]},{"constructor":23,"fields":[]}]}'
+    )
+    with pytest.raises(DeserializeException):
+        VestingParam.from_json(test2)
+
+
+def test_plutus_data_from_json_wrong_data_structure():
+    test = (
+        '{"constructor": 130, "fields": [{"int": 123}, {"bytes": "31323334"}, '
+        '{"wrong_list": [{"int": 4}, {"int": 5}, {"int": 6}]}, {"map": [{"v": {"bytes": "31"}, '
+        '"k": {"int": 1}}, {"v": {"bytes": "32"}, "k": {"int": 2}}]}]}'
+    )
+    with pytest.raises(DeserializeException):
+        MyTest.from_json(test)
+
+
+def test_plutus_data_from_json_wrong_data_structure_type():
+    test = (
+        '[{"constructor": 130, "fields": [{"int": 123}, {"bytes": "31323334"}, '
+        '{"list": [{"int": 4}, {"int": 5}, {"int": 6}]}, {"map": [{"v": {"bytes": "31"}, '
+        '"k": {"int": 1}}, {"v": {"bytes": "32"}, "k": {"int": 2}}]}]}]'
+    )
+    with pytest.raises(TypeError):
+        MyTest.from_json(test)
+
+
 def test_plutus_data_hash():
     assert (
         bytes.fromhex(
