From fff75197bfd6e1139e4c53a10d146342fb77e7f8 Mon Sep 17 00:00:00 2001 From: Julius Frost <33183774+juliusfrost@users.noreply.github.com> Date: Wed, 22 Mar 2023 21:31:58 -0400 Subject: [PATCH 1/2] utxo_by_tx_id --- pycardano/backend/ogmios.py | 112 ++++++++++++++++++++---------------- 1 file changed, 63 insertions(+), 49 deletions(-) diff --git a/pycardano/backend/ogmios.py b/pycardano/backend/ogmios.py index db9ccd2a..d70eb33d 100644 --- a/pycardano/backend/ogmios.py +++ b/pycardano/backend/ogmios.py @@ -369,61 +369,75 @@ def _utxos_ogmios(self, address: str) -> List[UTxO]: utxos = [] for result in results: - in_ref = result[0] - output = result[1] - tx_in = TransactionInput.from_primitive([in_ref["txId"], in_ref["index"]]) + utxos.append(self._utxo_from_ogmios_result(result)) - lovelace_amount = output["value"]["coins"] - - script = output.get("script", None) - if script: - if "plutus:v2" in script: - script = PlutusV2Script( - cbor2.loads(bytes.fromhex(script["plutus:v2"])) - ) - elif "plutus:v1" in script: - script = PlutusV1Script( - cbor2.loads(bytes.fromhex(script["plutus:v1"])) - ) - else: - raise ValueError("Unknown plutus script type") + return utxos - datum_hash = ( - DatumHash.from_primitive(output["datumHash"]) - if output.get("datumHash", None) - else None - ) + def utxo_by_tx_id(self, tx_id: str, index: int) -> Optional[UTxO]: + """Get a UTxO associated with a tx_id and index. - datum = None + Args: + tx_id (str): The transaction id. + index (int): The index for the UTxO at the specified transaction. - if output["datum"] and output["datum"] != output["datumHash"]: - datum = RawCBOR(bytes.fromhex(output["datum"])) + Returns: + Optional[UTxO]: Return a UTxO if exists at the tx_id and index. + """ + results = self._query_utxos_by_tx_id(tx_id, index) - if not output["value"]["assets"]: - tx_out = TransactionOutput( - Address.from_primitive(address), - amount=lovelace_amount, - datum_hash=datum_hash, - datum=datum, - script=script, - ) + utxos = [] + for result in results: + utxos.append(self._utxo_from_ogmios_result(result)) + + if len(utxos) > 0: + return utxos[0] + return None + + def _utxo_from_ogmios_result(self, result) -> UTxO: + in_ref = result[0] + output = result[1] + tx_in = TransactionInput.from_primitive([in_ref["txId"], in_ref["index"]]) + lovelace_amount = output["value"]["coins"] + script = output.get("script", None) + if script: + if "plutus:v2" in script: + script = PlutusV2Script(cbor2.loads(bytes.fromhex(script["plutus:v2"]))) + elif "plutus:v1" in script: + script = PlutusV1Script(cbor2.loads(bytes.fromhex(script["plutus:v1"]))) else: - multi_assets = MultiAsset() - - for asset, quantity in output["value"]["assets"].items(): - policy_hex, policy, asset_name_hex = self._extract_asset_info(asset) - multi_assets.setdefault(policy, Asset())[asset_name_hex] = quantity - - tx_out = TransactionOutput( - Address.from_primitive(address), - amount=Value(lovelace_amount, multi_assets), - datum_hash=datum_hash, - datum=datum, - script=script, - ) - utxos.append(UTxO(tx_in, tx_out)) - - return utxos + raise ValueError("Unknown plutus script type") + datum_hash = ( + DatumHash.from_primitive(output["datumHash"]) + if output.get("datumHash", None) + else None + ) + datum = None + if output["datum"] and output["datum"] != output["datumHash"]: + datum = RawCBOR(bytes.fromhex(output["datum"])) + if not output["value"]["assets"]: + tx_out = TransactionOutput( + Address.from_primitive(output["address"]), + amount=lovelace_amount, + datum_hash=datum_hash, + datum=datum, + script=script, + ) + else: + multi_assets = MultiAsset() + + for asset, quantity in output["value"]["assets"].items(): + policy_hex, policy, asset_name_hex = self._extract_asset_info(asset) + multi_assets.setdefault(policy, Asset())[asset_name_hex] = quantity + + tx_out = TransactionOutput( + Address.from_primitive(output["address"]), + amount=Value(lovelace_amount, multi_assets), + datum_hash=datum_hash, + datum=datum, + script=script, + ) + utxo = UTxO(tx_in, tx_out) + return utxo def submit_tx(self, cbor: Union[bytes, str]): """Submit a transaction to the blockchain. From 295a455f0d3fd19897528b43510aff230c6d439c Mon Sep 17 00:00:00 2001 From: Julius Frost <33183774+juliusfrost@users.noreply.github.com> Date: Mon, 27 Mar 2023 01:01:18 -0400 Subject: [PATCH 2/2] utxo_by_tx_id ogmios test --- pycardano/backend/ogmios.py | 1 + test/pycardano/backend/test_ogmios.py | 28 ++++++++++++++++++++++++++- 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/pycardano/backend/ogmios.py b/pycardano/backend/ogmios.py index d70eb33d..367db195 100644 --- a/pycardano/backend/ogmios.py +++ b/pycardano/backend/ogmios.py @@ -384,6 +384,7 @@ def utxo_by_tx_id(self, tx_id: str, index: int) -> Optional[UTxO]: Optional[UTxO]: Return a UTxO if exists at the tx_id and index. """ results = self._query_utxos_by_tx_id(tx_id, index) + assert len(results) < 2, f"Query for UTxO {tx_id}#{index} should be unique!" utxos = [] for result in results: diff --git a/test/pycardano/backend/test_ogmios.py b/test/pycardano/backend/test_ogmios.py index 8e7ac383..39afbd76 100644 --- a/test/pycardano/backend/test_ogmios.py +++ b/test/pycardano/backend/test_ogmios.py @@ -89,7 +89,17 @@ def override_request(method, args): elif args["query"] == "genesisConfig": return GENESIS_RESULT elif "utxo" in args["query"]: - return UTXOS + query = args["query"]["utxo"][0] + if isinstance(query, dict): + for utxo in UTXOS: + if ( + utxo[0]["txId"] == query["txId"] + and utxo[0]["index"] == query["index"] + ): + return [utxo] + return [] + else: + return UTXOS else: return None @@ -174,3 +184,19 @@ def test_utxo(self): }, } ) + + def test_utxo_by_tx_id(self): + utxo = self.chain_context.utxo_by_tx_id( + "3a42f652bd8dee788577e8c39b6217db3df659c33b10a2814c20fb66089ca167", + 1, + ) + assert utxo.input == TransactionInput.from_primitive( + ["3a42f652bd8dee788577e8c39b6217db3df659c33b10a2814c20fb66089ca167", 1] + ) + assert utxo.output.amount == 764295183 + + not_utxo = self.chain_context.utxo_by_tx_id( + "3a42f652bd8dee788577e8c39b6217db3df659c33b10a2814c20fb66089ca167", + 2, + ) + assert not_utxo is None