Improve base + blockfrost module maintainability

integration-test/test/test_ogmios.py

@@ -0,0 +1,14 @@
+from retry import retry
+
+from .base import TEST_RETRIES, TestBase
+
+
+class TestProtocolParam(TestBase):
+    @retry(tries=TEST_RETRIES, backoff=1.5, delay=6, jitter=(0, 4))
+    def test_protocol_param_cost_models(self):
+        protocol_param = self.chain_context.protocol_param
+
+        cost_models = protocol_param.cost_models
+        for _, cost_model in cost_models.items():
+            assert "addInteger-cpu-arguments-intercept" in cost_model
+            assert "addInteger-cpu-arguments-slope" in cost_model

pycardano/backend/base.py

@@ -19,90 +19,90 @@
 ALONZO_COINS_PER_UTXO_WORD = 34482
 
 
-@dataclass
+@dataclass(frozen=True)
 class GenesisParameters:
     """Cardano genesis parameters"""
 
-    active_slots_coefficient: float = None
+    active_slots_coefficient: float
 
-    update_quorum: int = None
+    update_quorum: int
 
-    max_lovelace_supply: int = None
+    max_lovelace_supply: int
 
-    network_magic: int = None
+    network_magic: int
 
-    epoch_length: int = None
+    epoch_length: int
 
-    system_start: int = None
+    system_start: int
 
-    slots_per_kes_period: int = None
+    slots_per_kes_period: int
 
-    slot_length: int = None
+    slot_length: int
 
-    max_kes_evolutions: int = None
+    max_kes_evolutions: int
 
-    security_param: int = None
+    security_param: int
 
 
-@dataclass
+@dataclass(frozen=True)
 class ProtocolParameters:
     """Cardano protocol parameters"""
 
-    min_fee_constant: int = None
+    min_fee_constant: int
 
-    min_fee_coefficient: int = None
+    min_fee_coefficient: int
 
-    max_block_size: int = None
+    max_block_size: int
 
-    max_tx_size: int = None
+    max_tx_size: int
 
-    max_block_header_size: int = None
+    max_block_header_size: int
 
-    key_deposit: int = None
+    key_deposit: int
 
-    pool_deposit: int = None
+    pool_deposit: int
 
-    pool_influence: float = None
+    pool_influence: float
 
-    monetary_expansion: float = None
+    monetary_expansion: float
 
-    treasury_expansion: float = None
+    treasury_expansion: float
 
-    decentralization_param: float = None
+    decentralization_param: float
 
-    extra_entropy: str = None
+    extra_entropy: str
 
-    protocol_major_version: int = None
+    protocol_major_version: int
 
-    protocol_minor_version: int = None
+    protocol_minor_version: int
 
-    min_utxo: int = None
+    min_utxo: int
 
-    min_pool_cost: int = None
+    min_pool_cost: int
 
-    price_mem: float = None
+    price_mem: float
 
-    price_step: float = None
+    price_step: float
 
-    max_tx_ex_mem: int = None
+    max_tx_ex_mem: int
 
-    max_tx_ex_steps: int = None
+    max_tx_ex_steps: int
 
-    max_block_ex_mem: int = None
+    max_block_ex_mem: int
 
-    max_block_ex_steps: int = None
+    max_block_ex_steps: int
 
-    max_val_size: int = None
+    max_val_size: int
 
-    collateral_percent: int = None
+    collateral_percent: int
 
-    max_collateral_inputs: int = None
+    max_collateral_inputs: int
 
-    coins_per_utxo_word: int = None
+    coins_per_utxo_word: int
 
-    coins_per_utxo_byte: int = None
+    coins_per_utxo_byte: int
 
-    cost_models: Dict[str, Dict[str, int]] = None
+    cost_models: Dict[str, Dict[str, int]]
     """A dict contains cost models for Plutus. The key will be "PlutusV1", "PlutusV2", etc.
     The value will be a dict of cost model parameters."""
 

pycardano/backend/blockfrost.py

@@ -1,10 +1,11 @@
 import os
 import tempfile
 import time
-from typing import Dict, List, Union
+from typing import Dict, List, Optional, Union
 
 import cbor2
 from blockfrost import ApiUrls, BlockFrostApi
+from blockfrost.utils import Namespace
 
 from pycardano.address import Address
 from pycardano.backend.base import (
@@ -28,6 +29,7 @@
     UTxO,
     Value,
 )
+from pycardano.types import JsonDict
 
 __all__ = ["BlockFrostChainContext"]
 
@@ -40,6 +42,12 @@ class BlockFrostChainContext(ChainContext):
         network (Network): Network to use.
     """
 
+    api: BlockFrostApi
+    _epoch_info: Namespace
+    _epoch: Optional[int] = None
+    _genesis_param: Optional[GenesisParameters] = None
+    _protocol_param: Optional[ProtocolParameters] = None
+
     def __init__(
         self, project_id: str, network: Network = Network.TESTNET, base_url: str = None
     ):
@@ -72,7 +80,8 @@ def network(self) -> Network:
     @property
     def epoch(self) -> int:
         if not self._epoch or self._check_epoch_and_update():
-            self._epoch = self.api.epoch_latest().epoch
+            new_epoch: int = self.api.epoch_latest().epoch
+            self._epoch = new_epoch
         return self._epoch
 
     @property
@@ -107,6 +116,7 @@ def protocol_param(self) -> ProtocolParameters:
                 protocol_major_version=int(params.protocol_major_ver),
                 protocol_minor_version=int(params.protocol_minor_ver),
                 min_utxo=int(params.min_utxo),
+                min_pool_cost=int(params.min_pool_cost),
                 price_mem=float(params.price_mem),
                 price_step=float(params.price_step),
                 max_tx_ex_mem=int(params.max_tx_ex_mem),
@@ -138,9 +148,10 @@ def _get_script(
                 cbor2.loads(bytes.fromhex(self.api.script_cbor(script_hash).cbor))
             )
         else:
-            return NativeScript.from_dict(
-                self.api.script_json(script_hash, return_type="json")["json"]
-            )
+            script_json: JsonDict = self.api.script_json(
+                script_hash, return_type="json"
+            )["json"]
+            return NativeScript.from_dict(script_json)
 
     def utxos(self, address: str) -> List[UTxO]:
         results = self.api.address_utxos(address, gather_pages=True)
@@ -152,7 +163,7 @@ def utxos(self, address: str) -> List[UTxO]:
                 [result.tx_hash, result.output_index]
             )
             amount = result.amount
-            lovelace_amount = None
+            lovelace_amount = 0
             multi_assets = MultiAsset()
             for item in amount:
                 if item.unit == "lovelace":

pycardano/backend/ogmios.py

@@ -29,13 +29,11 @@
     UTxO,
     Value,
 )
+from pycardano.types import JsonDict
 
 __all__ = ["OgmiosChainContext"]
 
 
-JSON = Dict[str, Any]
-
-
 class OgmiosQueryType(str, Enum):
     Query = "Query"
     SubmitTx = "SubmitTx"
@@ -66,7 +64,7 @@ def __init__(
         self._genesis_param = None
         self._protocol_param = None
 
-    def _request(self, method: OgmiosQueryType, args: JSON) -> Any:
+    def _request(self, method: OgmiosQueryType, args: JsonDict) -> Any:
         ws = websocket.WebSocket()
         ws.connect(self._ws_url)
         request = json.dumps(
@@ -88,27 +86,27 @@ def _request(self, method: OgmiosQueryType, args: JSON) -> Any:
             )
         return json.loads(response)["result"]
 
-    def _query_current_protocol_params(self) -> JSON:
+    def _query_current_protocol_params(self) -> JsonDict:
         args = {"query": "currentProtocolParameters"}
         return self._request(OgmiosQueryType.Query, args)
 
-    def _query_genesis_config(self) -> JSON:
+    def _query_genesis_config(self) -> JsonDict:
         args = {"query": "genesisConfig"}
         return self._request(OgmiosQueryType.Query, args)
 
     def _query_current_epoch(self) -> int:
         args = {"query": "currentEpoch"}
         return self._request(OgmiosQueryType.Query, args)
 
-    def _query_chain_tip(self) -> JSON:
+    def _query_chain_tip(self) -> JsonDict:
         args = {"query": "chainTip"}
         return self._request(OgmiosQueryType.Query, args)
 
-    def _query_utxos_by_address(self, address: str) -> List[List[JSON]]:
+    def _query_utxos_by_address(self, address: str) -> List[List[JsonDict]]:
         args = {"query": {"utxo": [address]}}
         return self._request(OgmiosQueryType.Query, args)
 
-    def _query_utxos_by_tx_id(self, tx_id: str, index: int) -> List[List[JSON]]:
+    def _query_utxos_by_tx_id(self, tx_id: str, index: int) -> List[List[JsonDict]]:
         args = {"query": {"utxo": [{"txId": tx_id, "index": index}]}}
         return self._request(OgmiosQueryType.Query, args)
 
@@ -151,6 +149,7 @@ def _fetch_protocol_param(self) -> ProtocolParameters:
             extra_entropy=result.get("extraEntropy", ""),
             protocol_major_version=result["protocolVersion"]["major"],
             protocol_minor_version=result["protocolVersion"]["minor"],
+            min_utxo=self._get_min_utxo(),
             min_pool_cost=result["minPoolCost"],
             price_mem=self._fraction_parser(result["prices"]["memory"]),
             price_step=self._fraction_parser(result["prices"]["steps"]),
@@ -165,17 +164,24 @@ def _fetch_protocol_param(self) -> ProtocolParameters:
                 "coinsPerUtxoWord", ALONZO_COINS_PER_UTXO_WORD
             ),
             coins_per_utxo_byte=result.get("coinsPerUtxoByte", 0),
-            cost_models=result.get("costModels", {}),
+            cost_models=self._parse_cost_models(result),
         )
 
-        if "plutus:v1" in param.cost_models:
-            param.cost_models["PlutusV1"] = param.cost_models.pop("plutus:v1")
-        if "plutus:v2" in param.cost_models:
-            param.cost_models["PlutusV2"] = param.cost_models.pop("plutus:v2")
+        return param
 
+    def _get_min_utxo(self) -> int:
         result = self._query_genesis_config()
-        param.min_utxo = result["protocolParameters"]["minUtxoValue"]
-        return param
+        return result["protocolParameters"]["minUtxoValue"]
+
+    def _parse_cost_models(self, ogmios_result: JsonDict) -> Dict[str, Dict[str, int]]:
+        ogmios_cost_models = ogmios_result.get("costModels", {})
+
+        cost_models = {}
+        if "plutus:v1" in ogmios_cost_models:
+            cost_models["PlutusV1"] = ogmios_cost_models["plutus:v1"].copy()
+        if "plutus:v2" in ogmios_cost_models:
+            cost_models["PlutusV2"] = ogmios_cost_models["plutus:v2"].copy()
+        return cost_models
 
     @property
     def genesis_param(self) -> GenesisParameters: