Add Transaction Metadata CBOR Retrieval Functionality

pycardano/backend/__init__.py

@@ -3,5 +3,6 @@
 from .base import *
 from .blockfrost import *
 from .cardano_cli import *
+from .kupo import *
 from .ogmios_v5 import *
 from .ogmios_v6 import *

pycardano/backend/base.py

@@ -11,6 +11,7 @@
 from pycardano.plutus import ExecutionUnits
 from pycardano.transaction import Transaction, UTxO
 from pycardano.types import typechecked
+from pycardano.serialization import RawCBOR
 
 __all__ = [
     "GenesisParameters",
@@ -218,3 +219,20 @@ def evaluate_tx_cbor(self, cbor: Union[bytes, str]) -> Dict[str, ExecutionUnits]
             List[ExecutionUnits]: A list of execution units calculated for each of the transaction's redeemers
         """
         raise NotImplementedError()
+
+    def tx_metadata_cbor(
+        self, tx_id: str, slot: Optional[int] = None
+    ) -> Optional[RawCBOR]:
+        """Get metadata CBOR for a transaction.
+
+        Args:
+            tx_id (str): Transaction id for metadata to query.
+            slot (Optional[int]): Slot number. Required for some backends (e.g., Kupo).
+
+        Returns:
+            Optional[RawCBOR]: Metadata CBOR if found, None otherwise.
+
+        Raises:
+            NotImplementedError: If the method is not implemented in the subclass.
+        """
+        raise NotImplementedError("tx_metadata_cbor is not implemented in the subclass")

pycardano/backend/blockfrost.py

@@ -322,3 +322,29 @@ def evaluate_tx_cbor(self, cbor: Union[bytes, str]) -> Dict[str, ExecutionUnits]
                     getattr(result.EvaluationResult, k).steps,
                 )
             return return_val
+
+    def tx_metadata_cbor(
+        self, tx_id: str, slot: Optional[int] = None
+    ) -> Optional[RawCBOR]:
+        """Get metadata CBOR for a transaction using BlockFrost.
+
+        Args:
+            tx_id (str): Transaction id for metadata to query.
+            slot (Optional[int]): Slot number (not used in BlockFrost implementation).
+
+        Returns:
+            Optional[RawCBOR]: Metadata CBOR if found, None otherwise.
+
+        Raises:
+            :class:`ApiError`: When fails to get metadata.
+        """
+        try:
+            response = self.api.transaction_metadata_cbor(tx_id)
+            if response:
+                return RawCBOR(bytes.fromhex(response[0].metadata))
+            return None
+        except ApiError as e:
+            if e.status_code == 404:
+                return None
+            else:
+                raise e

pycardano/backend/kupo.py

@@ -49,6 +49,7 @@ class KupoChainContextExtension(ChainContext):
     _kupo_url: Optional[str]
     _utxo_cache: Cache
     _datum_cache: Cache
+    _metadata_cache: Cache
     _refetch_chain_tip_interval: int
 
     def __init__(
@@ -58,6 +59,7 @@ def __init__(
         refetch_chain_tip_interval: int = 10,
         utxo_cache_size: int = 1000,
         datum_cache_size: int = 1000,
+        metadata_cache_size: int = 1000,
     ):
         self._kupo_url = kupo_url
         self._wrapped_backend = wrapped_backend
@@ -66,6 +68,7 @@ def __init__(
             ttl=self._refetch_chain_tip_interval, maxsize=utxo_cache_size
         )
         self._datum_cache = LRUCache(maxsize=datum_cache_size)
+        self._metadata_cache = LRUCache(maxsize=metadata_cache_size)
 
     @property
     def genesis_param(self) -> GenesisParameters:
@@ -253,3 +256,47 @@ 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 tx_metadata_cbor(
+        self, tx_id: str, slot: Optional[int] = None
+    ) -> Optional[RawCBOR]:
+        """Get metadata CBOR from Kupo or fallback to wrapped backend.
+
+        Args:
+            tx_id (str): Transaction id for metadata to query.
+            slot (Optional[int]): Slot number. Required for Kupo backend.
+
+        Returns:
+            Optional[RawCBOR]: Metadata CBOR if found, None otherwise.
+        """
+        if self._kupo_url is None:
+            raise AssertionError(
+                "kupo_url object attribute has not been assigned properly."
+            )
+
+        if slot is None:
+            raise ValueError("Slot number is required for Kupo backend.")
+
+        cache_key = (tx_id, slot)
+        if cache_key in self._metadata_cache:
+            return self._metadata_cache[cache_key]
+
+        kupo_metadata_url = f"{self._kupo_url}/metadata/{slot}?transaction_id={tx_id}"
+
+        try:
+            response = requests.get(kupo_metadata_url, timeout=10)
+            response.raise_for_status()
+            metadata_result = response.json()
+
+            if metadata_result and "raw" in metadata_result[0]:
+                metadata = RawCBOR(bytes.fromhex(metadata_result[0]["raw"]))
+                self._metadata_cache[cache_key] = metadata
+                return metadata
+
+            return None
+
+        except requests.exceptions.HTTPError as e:
+            if e.response.status_code == 404:
+                return None
+
+            raise

pycardano/backend/ogmios_v6.py

@@ -356,6 +356,25 @@ def _parse_cost_models(self, plutus_cost_models):
                 cost_models["PlutusV3"][f"{i:0{width}d}"] = v
         return cost_models
 
+    def tx_metadata_cbor(
+        self, tx_id: str, slot: Optional[int] = None
+    ) -> Optional[RawCBOR]:
+        """Get metadata CBOR for a transaction using Ogmios.
+
+        Args:
+            tx_id (str): Transaction id for metadata to query.
+            slot (Optional[int]): Slot number (not used in Ogmios implementation).
+
+        Returns:
+            Optional[RawCBOR]: Metadata CBOR if found, None otherwise.
+
+        Raises:
+            NotImplementedError: This method is not yet implemented for Ogmios V6 backend.
+        """
+        raise NotImplementedError(
+            "get_metadata_cbor is not yet implemented for Ogmios V6 backend"
+        )
+
 
 class OgmiosChainContext(OgmiosV6ChainContext):
     """An alias of OgmiosV6ChainContext for backwards compatibility."""
@@ -373,6 +392,7 @@ def KupoOgmiosV6ChainContext(
     network: Network = Network.TESTNET,
     kupo_url: Optional[str] = None,
 ) -> KupoChainContextExtension:
+    """Create a KupoChainContextExtension with an OgmiosV6ChainContext backend and Kupo URL."""
     return KupoChainContextExtension(
         OgmiosV6ChainContext(
             host,