Enable UTxO query with Kupo (#39)

* incorporate split change logic

* incorporate split change logic

* incorporate split change logic

* minor modification to change combine logic

* minor modification to change combine logic

* minor modification to change combine logic

* integrate diff changes

* integrate diff changes and remove unused functions

* add change split test case. Modify txbuilder logic in handeling splits.

* Bump pytest from 7.0.1 to 7.1.1

Bumps [pytest](https://github.com/pytest-dev/pytest) from 7.0.1 to 7.1.1.
- [Release notes](https://github.com/pytest-dev/pytest/releases)
- [Changelog](https://github.com/pytest-dev/pytest/blob/main/CHANGELOG.rst)
- [Commits](pytest-dev/pytest@7.0.1...7.1.1)

---
updated-dependencies:
- dependency-name: pytest
  dependency-type: direct:development
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <[email protected]>

* replace max fee with more accurately estimated fees

* replace max fee with more precise fee estimation

* remove max fee import

* modify format to be consistent with code base

* supports utxo queries using kupo

* support utxo query with Kupo

* reformat

* query utxo with Kupo when kupo url is provided

* remove the hardcoded genesis utxo in test_all file

Co-authored-by: Yuanheng Wang <[email protected]>
Co-authored-by: Yuanheng Wang <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Author: 4 people

integration-test/docker-compose.yml

@@ -58,6 +58,25 @@ services:
         max-size: "200k"
         max-file: "10"
 
+  kupo:
+    image: cardanosolutions/kupo
+    environment:
+      NETWORK: "${NETWORK:-local}"
+
+    command: [
+      "--node-socket", "/ipc/node.socket",
+      "--node-config", "/code/configs/${NETWORK:-local}/config.json",
+      "--host", "0.0.0.0",
+      "--since", "origin",
+      "--match", "*",
+      "--in-memory"
+    ]
+    volumes:
+      - .:/code
+      - node-ipc:/ipc
+    ports:
+      - ${KUPO_PORT:-1442}:1442
+
 volumes:
   node-db:
   node-ipc:

integration-test/run_tests.sh

@@ -26,4 +26,4 @@ sleep 8
 poetry run pytest -s -n 4 "$ROOT"/test
 
 # Cleanup
-docker-compose down --volumes --remove-orphans
+docker-compose down --volumes --remove-orphans

integration-test/test/test_all.py

@@ -22,7 +22,9 @@ class TestAll:
 
     OGMIOS_WS = "ws://localhost:1337"
 
-    chain_context = OgmiosChainContext(OGMIOS_WS, Network.TESTNET)
+    KUPO_URL = "http://localhost:1442/v1/matches"
+
+    chain_context = OgmiosChainContext(OGMIOS_WS, Network.TESTNET, kupo_url=KUPO_URL)
 
     check_chain_context(chain_context)
 
@@ -214,7 +216,7 @@ def test_plutus(self):
 
         # ----------- Giver give ---------------
 
-        with open("plutus_scripts/fortytwo.plutus", "r") as f:
+        with open("./plutus_scripts/fortytwo.plutus", "r") as f:
             script_hex = f.read()
             forty_two_script = cbor2.loads(bytes.fromhex(script_hex))
 

pycardano/backend/ogmios.py

@@ -3,6 +3,7 @@
 import time
 from typing import Dict, List, Union
 
+import requests
 import websocket
 
 from pycardano.address import Address
@@ -25,10 +26,17 @@
 
 
 class OgmiosChainContext(ChainContext):
-    def __init__(self, ws_url: str, network: Network, compact_result=True):
+    def __init__(
+        self,
+        ws_url: str,
+        network: Network,
+        compact_result=True,
+        kupo_url=None,
+    ):
         self._ws_url = ws_url
         self._network = network
         self._service_name = "ogmios.v1:compact" if compact_result else "ogmios"
+        self._kupo_url = kupo_url
         self._last_known_block_slot = 0
         self._genesis_param = None
         self._protocol_param = None
@@ -152,15 +160,103 @@ def last_block_slot(self) -> int:
         args = {"query": "chainTip"}
         return self._request(method, args)["slot"]
 
-    def utxos(self, address: str) -> List[UTxO]:
-        """Get all UTxOs associated with an address.
+    def _extract_asset_info(self, asset_hash: str):
+        policy_hex, asset_name_hex = asset_hash.split(".")
+        policy = ScriptHash.from_primitive(policy_hex)
+        asset_name_hex = AssetName.from_primitive(asset_name_hex)
+
+        return policy_hex, policy, asset_name_hex
+
+    def _check_utxo_unspent(self, tx_id: str, index: int) -> bool:
+        """Check whether an UTxO is unspent with Ogmios.
+
+        Args:
+            tx_id (str): transaction id.
+            index (int): transaction index.
+        """
+
+        method = "Query"
+        args = {"query": {"utxo": [{"txId": tx_id, "index": index}]}}
+        results = self._request(method, args)
+
+        if results:
+            return True
+        else:
+            return False
+
+    def _utxos_kupo(self, address: str) -> List[UTxO]:
+        """Get all UTxOs associated with an address with Kupo.
+        Since UTxO querying will be deprecated from Ogmios in next
+        major release: https://ogmios.dev/mini-protocols/local-state-query/.
+
+        Args:
+            address (str): An address encoded with bech32.
+
+        Returns:
+            List[UTxO]: A list of UTxOs.
+        """
+        address_url = self._kupo_url + "/" + address
+        results = requests.get(address_url).json()
+
+        utxos = []
+
+        for result in results:
+            tx_id = result["transaction_id"]
+            index = result["output_index"]
+
+            # Right now, all UTxOs of the address will be returned with Kupo, which requires Ogmios to
+            # validate if the UTxOs are spent with output reference. This feature is being considered to
+            # be added to Kupo to avoid extra API calls.
+            # See discussion here: https://github.com/CardanoSolutions/kupo/discussions/19.
+            if self._check_utxo_unspent(tx_id, index):
+                tx_in = TransactionInput.from_primitive([tx_id, index])
+
+                lovelace_amount = result["value"]["coins"]
+
+                datum_hash = (
+                    DatumHash.from_primitive(result["datum_hash"])
+                    if result["datum_hash"]
+                    else None
+                )
+
+                if not result["value"]["assets"]:
+                    tx_out = TransactionOutput(
+                        Address.from_primitive(address),
+                        amount=lovelace_amount,
+                        datum_hash=datum_hash,
+                    )
+                else:
+                    multi_assets = MultiAsset()
+
+                    for asset, quantity in result["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,
+                    )
+                utxos.append(UTxO(tx_in, tx_out))
+            else:
+                continue
+
+        return utxos
+
+    def _utxos_ogmios(self, address: str) -> List[UTxO]:
+        """Get all UTxOs associated with an address with Ogmios.
 
         Args:
             address (str): An address encoded with bech32.
 
         Returns:
             List[UTxO]: A list of UTxOs.
         """
+
         method = "Query"
         args = {"query": {"utxo": [address]}}
         results = self._request(method, args)
@@ -187,11 +283,8 @@ def utxos(self, address: str) -> List[UTxO]:
             else:
                 multi_assets = MultiAsset()
 
-                for asset in output["value"]["assets"]:
-                    quantity = output["value"]["assets"][asset]
-                    policy_hex, asset_name_hex = asset.split(".")
-                    policy = ScriptHash.from_primitive(policy_hex)
-                    asset_name_hex = AssetName.from_primitive(asset_name_hex)
+                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(
@@ -203,6 +296,22 @@ def utxos(self, address: str) -> List[UTxO]:
 
         return utxos
 
+    def utxos(self, address: str) -> List[UTxO]:
+        """Get all UTxOs associated with an address.
+
+        Args:
+            address (str): An address encoded with bech32.
+
+        Returns:
+            List[UTxO]: A list of UTxOs.
+        """
+        if self._kupo_url:
+            utxos = self._utxos_kupo(address)
+        else:
+            utxos = self._utxos_ogmios(address)
+
+        return utxos
+
     def submit_tx(self, cbor: Union[bytes, str]):
         """Submit a transaction to the blockchain.
 

pycardano/txbuilder.py

@@ -738,6 +738,7 @@ def build(
             )
         else:
             unfulfilled_amount.coin = max(0, unfulfilled_amount.coin)
+
         # Clean up all non-positive assets
         unfulfilled_amount.multi_asset = unfulfilled_amount.multi_asset.filter(
             lambda p, n, v: v > 0

test/pycardano/test_txbuilder.py

@@ -175,6 +175,7 @@ def test_tx_builder_raises_utxo_selection(chain_context):
 
     with pytest.raises(UTxOSelectionException) as e:
         tx_body = tx_builder.build(change_address=sender_address)
+
     # The unfulfilled amount includes requested (991000000) and estimated fees (161277)
     assert "Unfulfilled amount:\n {'coin': 991161277" in e.value.args[0]
     assert "{AssetName(b'NewToken'): 1}" in e.value.args[0]