diff --git a/pycardano/backend/__init__.py b/pycardano/backend/__init__.py index 617d36c1..0a384125 100644 --- a/pycardano/backend/__init__.py +++ b/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 * diff --git a/pycardano/backend/base.py b/pycardano/backend/base.py index 7e3fb145..f8831c28 100644 --- a/pycardano/backend/base.py +++ b/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") diff --git a/pycardano/backend/blockfrost.py b/pycardano/backend/blockfrost.py index 3d5d1afe..98af9bd3 100644 --- a/pycardano/backend/blockfrost.py +++ b/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 diff --git a/pycardano/backend/kupo.py b/pycardano/backend/kupo.py index 36fa409e..af9decf4 100644 --- a/pycardano/backend/kupo.py +++ b/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 diff --git a/pycardano/backend/ogmios_v6.py b/pycardano/backend/ogmios_v6.py index 7e60f845..8e41c68d 100644 --- a/pycardano/backend/ogmios_v6.py +++ b/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,