Add missing methods for mempool, add tx_submit_cbor and tx_evaluate_cbor accepting cbor bytes/string

blockfrost/api/__init__.py

@@ -95,6 +95,10 @@ def root(self, **kwargs):
         epoch_protocol_parameters
     from .cardano.ledger import \
         genesis
+    from .cardano.mempool import \
+        mempool, \
+        mempool_address, \
+        mempool_tx
     from .cardano.metadata import \
         metadata_labels, \
         metadata_label_json, \
@@ -125,8 +129,11 @@ def root(self, **kwargs):
         transaction_metadata, \
         transaction_metadata_cbor, \
         transaction_submit, \
+        transaction_submit_cbor, \
         transaction_redeemers, \
-        transaction_evaluate
+        transaction_evaluate, \
+        transaction_evaluate_cbor, \
+        transaction_evaluate_utxos
     from .cardano.scripts import \
         scripts, \
         script, \

blockfrost/api/cardano/mempool.py

@@ -0,0 +1,88 @@
+import requests
+from blockfrost.utils import list_request_wrapper
+
+
+@list_request_wrapper
+def mempool(self, **kwargs):
+    """
+    Obtains transactions that are currently stored in Blockfrost mempool, waiting to be included in a newly minted block.
+    Returns only transactions submitted via Blockfrost.io.
+
+    https://docs.blockfrost.io/#tag/Cardano-Mempool/paths/~1mempool/get
+
+    :param return_type: Optional. "object", "json" or "pandas". Default: "object".
+    :type return_type: str
+    :param gather_pages: Optional. Default: false. Will collect all pages into one return
+    :type gather_pages: bool
+    :param count: Optional. Default: 100. The number of results displayed on one page.
+    :type count: int
+    :param page: Optional. The page number for listing the results.
+    :type page: int
+    :returns A list of objects.
+    :rtype [Namespace]
+    :raises ApiError: If API fails
+    :raises Exception: If the API response is somehow malformed.
+    """
+    return requests.get(
+        url=f"{self.url}/mempool",
+        params=self.query_parameters(kwargs),
+        headers=self.default_headers
+    )
+
+@list_request_wrapper
+def mempool_tx(self, hash: str, **kwargs):
+    """
+    Obtains mempool transaction
+
+    https://docs.blockfrost.io/#tag/Cardano-Mempool/paths/~1mempool~1%7Bhash%7D/get
+
+    :param hash: Hash of the requested transaction.
+    :type hash: str
+    :param return_type: Optional. "object", "json" or "pandas". Default: "object".
+    :type return_type: str
+    :param gather_pages: Optional. Default: false. Will collect all pages into one return
+    :type gather_pages: bool
+    :param count: Optional. Default: 100. The number of results displayed on one page.
+    :type count: int
+    :param page: Optional. The page number for listing the results.
+    :type page: int
+    :returns A list of objects.
+    :rtype [Namespace]
+    :raises ApiError: If API fails
+    :raises Exception: If the API response is somehow malformed.
+    """
+    return requests.get(
+        url=f"{self.url}/mempool/{hash}",
+        params=self.query_parameters(kwargs),
+        headers=self.default_headers
+    )
+
+@list_request_wrapper
+def mempool_address(self, address: str, **kwargs):
+    """
+    Obtains list of mempool transactions where at least one of the transaction inputs or outputs belongs to the address (paginated).
+    Shows only transactions submitted via Blockfrost.io.
+
+    https://docs.blockfrost.io/#tag/Cardano-Mempool/paths/~1mempool~1addresses~1%7Baddress%7D/get
+
+    :param address: Bech32 address.
+    :type address: str
+    :param return_type: Optional. "object", "json" or "pandas". Default: "object".
+    :type return_type: str
+    :param gather_pages: Optional. Default: false. Will collect all pages into one return
+    :type gather_pages: bool
+    :param count: Optional. Default: 100. The number of results displayed on one page.
+    :type count: int
+    :param page: Optional. The page number for listing the results.
+    :type page: int
+    :returns A list of objects.
+    :rtype [Namespace]
+    :raises ApiError: If API fails
+    :raises Exception: If the API response is somehow malformed.
+    """
+    return requests.get(
+        url=f"{self.url}/mempool/addresses/{address}",
+        params=self.query_parameters(kwargs),
+        headers=self.default_headers
+    )
+

blockfrost/api/cardano/transactions.py

@@ -1,4 +1,5 @@
 import requests
+from typing import Union
 from blockfrost.utils import request_wrapper, list_request_wrapper
 
 
@@ -268,6 +269,36 @@ def transaction_submit(self, file_path: str, **kwargs):
         )
 
 
+@request_wrapper
+def transaction_submit_cbor(self, tx_cbor: Union[bytes, str], **kwargs):
+    """
+    Submit an already serialized transaction to the network.
+
+    https://docs.blockfrost.io/#tag/Cardano-Transactions/paths/~1tx~1submit/post
+
+    :param tx_cbor: Transaction in CBOR format, either as a hex-encoded string or as bytes.
+    :type tx_cbor: Union[str, bytes]
+    :returns str object.
+    :rtype str
+    :raises ApiError: If API fails
+    :raises Exception: If the API response is somehow malformed.
+    """
+
+    # Convert to bytes
+    if isinstance(tx_cbor, str):
+        data = bytes.fromhex(tx_cbor)
+    else:
+        data = tx_cbor
+
+    header = self.default_headers
+    header['Content-Type'] = 'application/cbor'
+    return requests.post(
+        url=f"{self.url}/tx/submit",
+        headers=header,
+        data=data,
+    )
+
+
 @request_wrapper
 def transaction_evaluate(self, file_path: str, **kwargs):
     """
@@ -290,3 +321,74 @@ def transaction_evaluate(self, file_path: str, **kwargs):
             headers=header,
             data=file,
         )
+
+
+@request_wrapper
+def transaction_evaluate_cbor(self, tx_cbor: Union[bytes, str], **kwargs):
+    """
+    Submit an already serialized transaction to evaluate how much execution units it requires.
+
+    https://docs.blockfrost.io/#tag/Cardano-Utilities/paths/~1utils~1txs~1evaluate/post
+
+    :param tx_cbor: Transaction in CBOR format, either as a hex-encoded string or as bytes.
+    :type tx_cbor: Union[str, bytes]
+    :returns str object.
+    :rtype str
+    :raises ApiError: If API fails
+    :raises Exception: If the API response is somehow malformed.
+    """
+    header = self.default_headers
+    header['Content-Type'] = 'application/cbor'
+
+    # Convert bytes to hex
+    if isinstance(tx_cbor, bytes):
+        data = tx_cbor.hex()
+    else:
+        data = tx_cbor
+
+    return requests.post(
+        url=f"{self.url}/utils/txs/evaluate",
+        headers=header,
+        data=data,
+    )
+
+
+@request_wrapper
+def transaction_evaluate_utxos(self, tx_cbor: Union[bytes, str], additional_utxo_set: list, **kwargs):
+    """
+    Submits a transaction CBOR and additional utxo set to evaluate how much execution units it requires.
+
+    https://docs.blockfrost.io/#tag/Cardano-Utilities/paths/~1utils~1txs~1evaluate~1utxos/post
+    https://ogmios.dev/mini-protocols/local-tx-submission/#evaluatetx
+
+    :param tx_cbor: Transaction in CBOR format, either as a hex-encoded string or as bytes.
+    :type tx_cbor: Union[bytes, str]
+    :param additional_utxo_set: Additional UTXO as an array of tuples [TxIn, TxOut] https://ogmios.dev/mini-protocols/local-tx-submission/#additional-utxo-set.
+    :type additional_utxo_set: list
+    :returns: Result of Ogmios EvaluateTx
+    :rtype dict
+    :raises ApiError: If API fails
+    :raises Exception: If the API response is somehow malformed.
+    """
+
+    # Convert bytes to hex
+    if isinstance(tx_cbor, bytes):
+        data = tx_cbor.hex()
+    else:
+        data = tx_cbor
+
+    headers = {
+        'Content-type': 'application/json',
+        **self.default_headers
+    }
+
+    payload = {
+        'cbor': data,
+        'additionalUtxoSet': additional_utxo_set
+    }
+
+    return requests.post(
+        url=f"{self.url}/utils/txs/evaluate/utxos",
+        headers=headers,
+        json=payload
+    )

blockfrost/utils.py

@@ -47,6 +47,7 @@ def convert_json_to_pandas(json_response):
 
 
 def simple_request_wrapper(func):
+    @wraps(func)
     def error_wrapper(*args, **kwargs):
         request_response: Response = func(*args, **kwargs)
         if request_response.status_code != 200:
@@ -58,6 +59,7 @@ def error_wrapper(*args, **kwargs):
 
 
 def request_wrapper(func):
+    @wraps(func)
     def error_wrapper(*args, **kwargs):
         request_response: Response = func(*args, **kwargs)
         if request_response.status_code != 200:
@@ -77,6 +79,7 @@ def error_wrapper(*args, **kwargs):
 
 
 def list_request_wrapper(func):
+    @wraps(func)
     def pagination(*args, **kwargs):
         def recursive_append(json_list, *args, **kwargs):
             request_response: Response = func(*args, **kwargs)
@@ -128,7 +131,8 @@ def __init__(
     ):
         self.project_id = project_id if project_id else os.environ.get(
             'BLOCKFROST_PROJECT_ID')
-        self.api_version = api_version
+        self.api_version = api_version if api_version else os.environ.get('BLOCKFROST_API_VERSION',
+                                                                          default=DEFAULT_API_VERSION)
         self.base_url = base_url
 
     @property

tests/test_cardano_addresses.py

@@ -3,7 +3,7 @@
 from blockfrost import BlockFrostApi, ApiError
 from blockfrost.utils import convert_json_to_object
 
-address = 'addr1qyptln5t5s0mastzc9rksn6wdqp9ynt67ahw0nhzukar5keu7yzv8ay6qvmlywtgvt7exaxt783dxuzv03qal7muda5sl42hg6'
+address = 'addr1qxk49ptelk7uda7acrczz30a7fu778sax5aapa38nhmve3eu7yzv8ay6qvmlywtgvt7exaxt783dxuzv03qal7muda5srnx35s'
 stake_address = 'stake1ux3g2c9dx2nhhehyrezyxpkstartcqmu9hk63qgfkccw5rqttygt7'
 asset = 'f4988f549728dc76b58d7677849443caf6e5385dc67e6c25f6aa901e506978656c54696c653235'
 
@@ -32,7 +32,8 @@ def test_address(requests_mock):
 
 def test_integration_address():
     if os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'):
-        api = BlockFrostApi(project_id=os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'))
+        api = BlockFrostApi(project_id=os.getenv(
+            'BLOCKFROST_PROJECT_ID_MAINNET'))
         assert api.address(address=address)
 
 
@@ -58,13 +59,16 @@ def test_address_extended(requests_mock):
         "type": "shelley",
         "script": False
     }
-    requests_mock.get(f"{api.url}/addresses/{address}/extended", json=mock_data)
-    assert api.address_extended(address=address) == convert_json_to_object(mock_data)
+    requests_mock.get(
+        f"{api.url}/addresses/{address}/extended", json=mock_data)
+    assert api.address_extended(
+        address=address) == convert_json_to_object(mock_data)
 
 
 def test_integration_address_extended():
     if os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'):
-        api = BlockFrostApi(project_id=os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'))
+        api = BlockFrostApi(project_id=os.getenv(
+            'BLOCKFROST_PROJECT_ID_MAINNET'))
         assert api.address_extended(address=address)
 
 
@@ -95,12 +99,14 @@ def test_address_total(requests_mock):
         "tx_count": 12
     }
     requests_mock.get(f"{api.url}/addresses/{address}/total", json=mock_data)
-    assert api.address_total(address=address) == convert_json_to_object(mock_data)
+    assert api.address_total(
+        address=address) == convert_json_to_object(mock_data)
 
 
 def test_integration_address_total():
     if os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'):
-        api = BlockFrostApi(project_id=os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'))
+        api = BlockFrostApi(project_id=os.getenv(
+            'BLOCKFROST_PROJECT_ID_MAINNET'))
         assert api.address_total(address=address)
 
 
@@ -152,13 +158,15 @@ def test_address_utxos(requests_mock):
         }
     ]
     requests_mock.get(f"{api.url}/addresses/{address}/utxos", json=mock_data)
-    assert api.address_utxos(address=address) == convert_json_to_object(mock_data)
+    assert api.address_utxos(
+        address=address) == convert_json_to_object(mock_data)
 
 
 def test_integration_address_utxos():
     if os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'):
-        api = BlockFrostApi(project_id=os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'))
-        assert api.address_utxos(address=address)
+        api = BlockFrostApi(project_id=os.getenv(
+            'BLOCKFROST_PROJECT_ID_MAINNET'))
+        assert api.address_utxos(address=address) == []
 
 
 def test_address_utxos_asset(requests_mock):
@@ -205,13 +213,16 @@ def test_address_utxos_asset(requests_mock):
             "data_hash": None
         }
     ]
-    requests_mock.get(f"{api.url}/addresses/{address}/utxos/{asset}", json=mock_data)
-    assert api.address_utxos_asset(address=address, asset=asset) == convert_json_to_object(mock_data)
+    requests_mock.get(
+        f"{api.url}/addresses/{address}/utxos/{asset}", json=mock_data)
+    assert api.address_utxos_asset(
+        address=address, asset=asset) == convert_json_to_object(mock_data)
 
 
 def test_integration_address_utxos_asset():
     if os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'):
-        api = BlockFrostApi(project_id=os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'))
+        api = BlockFrostApi(project_id=os.getenv(
+            'BLOCKFROST_PROJECT_ID_MAINNET'))
         assert api.address_utxos_asset(address=address, asset=asset) == []
 
 
@@ -237,11 +248,14 @@ def test_address_transactions(requests_mock):
             "block_time": 1834505492
         }
     ]
-    requests_mock.get(f"{api.url}/addresses/{address}/transactions", json=mock_data)
-    assert api.address_transactions(address=address) == convert_json_to_object(mock_data)
+    requests_mock.get(
+        f"{api.url}/addresses/{address}/transactions", json=mock_data)
+    assert api.address_transactions(
+        address=address) == convert_json_to_object(mock_data)
 
 
 def test_integration_address_transactions():
     if os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'):
-        api = BlockFrostApi(project_id=os.getenv('BLOCKFROST_PROJECT_ID_MAINNET'))
+        api = BlockFrostApi(project_id=os.getenv(
+            'BLOCKFROST_PROJECT_ID_MAINNET'))
         assert api.address_transactions(address=address)