Support decoding list/indefinite list from cbor

.gitignore

@@ -9,5 +9,6 @@ coverage.xml
 # IDE
 .idea
 .code
+.venv
 /integration-test/.env
 /integration-test/tmp_configs/*

pycardano/backend/kupo.py

@@ -21,6 +21,7 @@
     MultiAsset,
     TransactionInput,
     TransactionOutput,
+    TransactionId,
     UTxO,
     Value,
 )
@@ -197,6 +198,8 @@ def _utxos_kupo(self, address: str) -> List[UTxO]:
                 )
                 if datum_hash and result.get("datum_type", "inline"):
                     datum = self._get_datum_from_kupo(result["datum_hash"])
+                    if datum:
+                        datum_hash = None
 
                 if not result["value"]["assets"]:
                     tx_out = TransactionOutput(
@@ -253,3 +256,27 @@ def evaluate_tx_cbor(self, cbor: Union[bytes, str]) -> Dict[str, ExecutionUnits]
             :class:`TransactionFailedException`: When fails to evaluate the transaction.
         """
         return self._wrapped_backend.evaluate_tx_cbor(cbor)
+
+    def get_metadata_cbor(
+        self, tx_id: TransactionId, slot: int
+    ) -> Optional[RawCBOR]:
+        """Get metadata cbor from Kupo.
+
+        Args:
+            tx_id (TransactionId): Transaction id for metadata to query.
+            slot (int): Slot number.
+
+        Returns:
+            Optional[RawCBOR]: Metadata cbor."""
+        if self._kupo_url is None:
+            raise AssertionError(
+                "kupo_url object attribute has not been assigned properly."
+            )
+
+        kupo_metadata_url = self._kupo_url + f"/metadata/{slot}?transaction_id={tx_id}"
+        metadata_result = requests.get(kupo_metadata_url).json()
+
+        if metadata_result and metadata_result[0]["transaction_id"] == tx_id:
+            return RawCBOR(bytes.fromhex(metadata_result[0]["raw"]))
+
+        return None

pycardano/backend/ogmios_v6.py

@@ -59,6 +59,7 @@ def __init__(
         self,
         host: str = "localhost",
         port: int = 1337,
+        path: str = "",
         secure: bool = False,
         refetch_chain_tip_interval: Optional[float] = None,
         utxo_cache_size: int = 10000,
@@ -67,6 +68,7 @@ def __init__(
     ):
         self.host = host
         self.port = port
+        self.path = path
         self.secure = secure
         self._network = network
         self._service_name = "ogmios"
@@ -86,26 +88,26 @@ def __init__(
         self._datum_cache = LRUCache(maxsize=datum_cache_size)
 
     def _query_current_era(self) -> OgmiosEra:
-        with OgmiosClient(self.host, self.port, self.secure) as client:
+        with OgmiosClient(self.host, self.port, self.path, self.secure) as client:
             return get_current_era(client)
 
     def _query_current_epoch(self) -> int:
-        with OgmiosClient(self.host, self.port, self.secure) as client:
+        with OgmiosClient(self.host, self.port, self.path, self.secure) as client:
             epoch, _ = client.query_epoch.execute()
             return epoch
 
     def _query_chain_tip(self) -> OgmiosTip:
-        with OgmiosClient(self.host, self.port, self.secure) as client:
+        with OgmiosClient(self.host, self.port, self.path, self.secure) as client:
             tip, _ = client.query_network_tip.execute()
             return tip
 
     def _query_utxos_by_address(self, address: Address) -> List[OgmiosUtxo]:
-        with OgmiosClient(self.host, self.port, self.secure) as client:
+        with OgmiosClient(self.host, self.port, self.path, self.secure) as client:
             utxos, _ = client.query_utxo.execute([address])
             return utxos
 
     def _query_utxos_by_tx_id(self, tx_id: str, index: int) -> List[OgmiosUtxo]:
-        with OgmiosClient(self.host, self.port, self.secure) as client:
+        with OgmiosClient(self.host, self.port, self.path, self.secure) as client:
             utxos, _ = client.query_utxo.execute(
                 [OgmiosTxOutputReference(tx_id, index)]
             )
@@ -135,7 +137,7 @@ def protocol_param(self) -> ProtocolParameters:
         return self._protocol_param
 
     def _fetch_protocol_param(self) -> ProtocolParameters:
-        with OgmiosClient(self.host, self.port, self.secure) as client:
+        with OgmiosClient(self.host, self.port, self.path, self.secure) as client:
             protocol_parameters, _ = client.query_protocol_parameters.execute()
             pyc_protocol_params = ProtocolParameters(
                 min_fee_constant=protocol_parameters.min_fee_constant.lovelace,
@@ -205,7 +207,7 @@ def genesis_param(self) -> GenesisParameters:
         return self._genesis_param  # type: ignore[return-value]
 
     def _fetch_genesis_param(self) -> OgmiosGenesisParameters:
-        with OgmiosClient(self.host, self.port, self.secure) as client:
+        with OgmiosClient(self.host, self.port, self.path, self.secure) as client:
             return OgmiosGenesisParameters(client, self._query_current_era())
 
     @property
@@ -311,13 +313,13 @@ def utxo_by_tx_id(self, tx_id: str, index: int) -> Optional[UTxO]:
     def submit_tx_cbor(self, cbor: Union[bytes, str]):
         if isinstance(cbor, bytes):
             cbor = cbor.hex()
-        with OgmiosClient(self.host, self.port, self.secure) as client:
+        with OgmiosClient(self.host, self.port, self.path, self.secure) as client:
             client.submit_transaction.execute(cbor)
 
     def evaluate_tx_cbor(self, cbor: Union[bytes, str]) -> Dict[str, ExecutionUnits]:
         if isinstance(cbor, bytes):
             cbor = cbor.hex()
-        with OgmiosClient(self.host, self.port, self.secure) as client:
+        with OgmiosClient(self.host, self.port, self.path, self.secure) as client:
             result, _ = client.evaluate_transaction.execute(cbor)
             result_dict = {}
             for res in result:

pycardano/hash.py

@@ -72,6 +72,9 @@ def __hash__(self):
     def payload(self) -> bytes:
         return self._payload
 
+    def to_shallow_primitive(self) -> bytes:
+        return self.payload
+
     def to_primitive(self) -> bytes:
         return self.payload
 

pycardano/serialization.py

@@ -5,12 +5,14 @@
 import re
 import typing
 from collections import OrderedDict, UserList, defaultdict
+from collections.abc import Sequence
 from copy import deepcopy
 from dataclasses import Field, dataclass, fields
 from datetime import datetime
 from decimal import Decimal
 from functools import wraps
 from inspect import isclass
+from io import BytesIO
 from typing import (
     Any,
     Callable,
@@ -24,7 +26,14 @@
     get_type_hints,
 )
 
-from cbor2 import CBOREncoder, CBORSimpleValue, CBORTag, dumps, loads, undefined
+from cbor2 import (
+    CBORDecoder,
+    CBOREncoder,
+    CBORSimpleValue,
+    CBORTag,
+    dumps,
+)
+from cbor2._types import UndefinedType, break_marker
 from frozendict import frozendict
 from frozenlist import FrozenList
 from pprintpp import pformat
@@ -86,6 +95,7 @@ class RawCBOR:
 Primitive = Union[
     bytes,
     bytearray,
+    ByteString,
     str,
     int,
     float,
@@ -98,6 +108,7 @@ class RawCBOR:
     dict,
     defaultdict,
     OrderedDict,
+    UndefinedType,
     datetime,
     re.Pattern,
     CBORSimpleValue,
@@ -113,6 +124,7 @@ class RawCBOR:
 PRIMITIVE_TYPES = (
     bytes,
     bytearray,
+    ByteString,
     str,
     int,
     float,
@@ -125,7 +137,7 @@ class RawCBOR:
     dict,
     defaultdict,
     OrderedDict,
-    type(undefined),
+    UndefinedType,
     datetime,
     re.Pattern,
     CBORSimpleValue,
@@ -169,6 +181,38 @@ def wrapper(cls, value: Primitive):
 CBORBase = TypeVar("CBORBase", bound="CBORSerializable")
 
 
+class IndefiniteDecoder(CBORDecoder):
+    def decode_array(self, subtype: int) -> Sequence[Any]:
+        # Major tag 4
+        length = self._decode_length(subtype, allow_indefinite=True)
+        if length is None:
+            # Indefinite length
+            items: list = []
+            if not self._immutable:
+                self.set_shareable(items)
+            while True:
+                value = self._decode()
+                if value is break_marker:
+                    break
+                else:
+                    items.append(value)
+
+            return IndefiniteList(items)
+        else:
+            return super().decode_array(subtype=subtype)
+
+    major_decoders: dict[int, Callable[[Any, int], Any]] = {
+        0: CBORDecoder.decode_uint,
+        1: CBORDecoder.decode_negint,
+        2: CBORDecoder.decode_bytestring,
+        3: CBORDecoder.decode_string,
+        4: decode_array,
+        5: CBORDecoder.decode_map,
+        6: CBORDecoder.decode_semantic,
+        7: CBORDecoder.decode_special,
+    }
+
+
 def default_encoder(
     encoder: CBOREncoder, value: Union[CBORSerializable, IndefiniteList]
 ):
@@ -235,7 +279,7 @@ class CBORSerializable:
         does not refer to itself, which could cause infinite loops.
     """
 
-    def to_shallow_primitive(self) -> Primitive:
+    def to_shallow_primitive(self) -> Union[Primitive, "CBORSerializable"]:
         """
         Convert the instance to a CBOR primitive. If the primitive is a container, e.g. list, dict, the type of
         its elements could be either a Primitive or a CBORSerializable.
@@ -479,9 +523,13 @@ def from_cbor(cls, payload: Union[str, bytes]) -> CBORSerializable:
             TestParent(3, Test(1, 2))
 
         """
-        if type(payload) is str:
+        if isinstance(payload, str):
             payload = bytes.fromhex(payload)
-        value = loads(payload)  # type: ignore
+
+        with BytesIO(payload) as fp:
+            value = IndefiniteDecoder(fp).decode()
+
+        # value = loads(payload)  # type: ignore
         return cls.from_primitive(value)
 
     def __repr__(self):
@@ -503,6 +551,7 @@ def _restore_dataclass_field(
 
     if "object_hook" in f.metadata:
         return f.metadata["object_hook"](v)
+
     return _restore_typed_primitive(f.type, v)
 
 
@@ -528,10 +577,14 @@ def _restore_typed_primitive(
             raise DeserializeException(
                 f"List types need exactly one type argument, but got {t_args}"
             )
-        t = t_args[0]
-        if not isinstance(v, list):
+        t_subtype = t_args[0]
+        if not isinstance(v, (list, IndefiniteList)):
             raise DeserializeException(f"Expected type list but got {type(v)}")
-        return IndefiniteList([_restore_typed_primitive(t, w) for w in v])
+        v_list = [_restore_typed_primitive(t_subtype, w) for w in v]
+        if t == IndefiniteList:
+            return IndefiniteList(v_list)
+        else:
+            return v_list
     elif isclass(t) and t == ByteString:
         if not isinstance(v, bytes):
             raise DeserializeException(f"Expected type bytes but got {type(v)}")
@@ -657,8 +710,10 @@ def to_shallow_primitive(self) -> Primitive:
         return primitives
 
     @classmethod
-    @limit_primitive_type(list, tuple)
-    def from_primitive(cls: Type[ArrayBase], values: Union[list, tuple]) -> ArrayBase:
+    @limit_primitive_type(list, tuple, IndefiniteList)
+    def from_primitive(
+        cls: Type[ArrayBase], values: Union[list, tuple, IndefiniteList]
+    ) -> ArrayBase:
         """Restore a primitive value to its original class type.
 
         Args: