diff --git a/Makefile b/Makefile index 4b3cbf67..e2195782 100644 --- a/Makefile +++ b/Makefile @@ -63,6 +63,7 @@ test-single: ## runs tests with "single" markers qa: ## runs static analyses poetry run flake8 pycardano poetry run mypy --install-types --non-interactive pycardano + poetry run black --check . format: ## runs code style and formatter poetry run isort . diff --git a/pycardano/address.py b/pycardano/address.py index cbe07fef..3905b1c3 100644 --- a/pycardano/address.py +++ b/pycardano/address.py @@ -9,7 +9,7 @@ from __future__ import annotations from enum import Enum -from typing import Union, Type +from typing import Type, Union from pycardano.crypto.bech32 import decode, encode from pycardano.exception import ( diff --git a/pycardano/backend/ogmios.py b/pycardano/backend/ogmios.py index 163a3e6d..4fa2a47c 100644 --- a/pycardano/backend/ogmios.py +++ b/pycardano/backend/ogmios.py @@ -2,7 +2,7 @@ import json import time from enum import Enum -from typing import Any, Dict, List, Optional, Union, Tuple +from typing import Any, Dict, List, Optional, Tuple, Union import cbor2 import requests @@ -88,7 +88,31 @@ def _request(self, method: OgmiosQueryType, args: JSON) -> Any: ) return json.loads(response)["result"] - def _check_chain_tip_and_update(self): + def _query_current_protocol_params(self) -> JSON: + args = {"query": "currentProtocolParameters"} + return self._request(OgmiosQueryType.Query, args) + + def _query_genesis_config(self) -> JSON: + 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: + args = {"query": "chainTip"} + return self._request(OgmiosQueryType.Query, args) + + def _query_utxos_by_address(self, address: str) -> List[List[JSON]]: + 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]]: + args = {"query": {"utxo": [{"txId": tx_id, "index": index}]}} + return self._request(OgmiosQueryType.Query, args) + + def _is_chain_tip_updated(self): slot = self.last_block_slot if self._last_known_block_slot != slot: self._last_known_block_slot = slot @@ -104,84 +128,84 @@ def _fraction_parser(fraction: str) -> float: @property def protocol_param(self) -> ProtocolParameters: """Get current protocol parameters""" - args = {"query": "currentProtocolParameters"} - if not self._protocol_param or self._check_chain_tip_and_update(): - result = self._request(OgmiosQueryType.Query, args) - param = ProtocolParameters( - min_fee_constant=result["minFeeConstant"], - min_fee_coefficient=result["minFeeCoefficient"], - max_block_size=result["maxBlockBodySize"], - max_tx_size=result["maxTxSize"], - max_block_header_size=result["maxBlockHeaderSize"], - key_deposit=result["stakeKeyDeposit"], - pool_deposit=result["poolDeposit"], - pool_influence=self._fraction_parser(result["poolInfluence"]), - monetary_expansion=self._fraction_parser(result["monetaryExpansion"]), - treasury_expansion=self._fraction_parser(result["treasuryExpansion"]), - decentralization_param=self._fraction_parser( - result.get("decentralizationParameter", "0/1") - ), - extra_entropy=result.get("extraEntropy", ""), - protocol_major_version=result["protocolVersion"]["major"], - protocol_minor_version=result["protocolVersion"]["minor"], - min_pool_cost=result["minPoolCost"], - price_mem=self._fraction_parser(result["prices"]["memory"]), - price_step=self._fraction_parser(result["prices"]["steps"]), - max_tx_ex_mem=result["maxExecutionUnitsPerTransaction"]["memory"], - max_tx_ex_steps=result["maxExecutionUnitsPerTransaction"]["steps"], - max_block_ex_mem=result["maxExecutionUnitsPerBlock"]["memory"], - max_block_ex_steps=result["maxExecutionUnitsPerBlock"]["steps"], - max_val_size=result["maxValueSize"], - collateral_percent=result["collateralPercentage"], - max_collateral_inputs=result["maxCollateralInputs"], - coins_per_utxo_word=result.get( - "coinsPerUtxoWord", ALONZO_COINS_PER_UTXO_WORD - ), - coins_per_utxo_byte=result.get("coinsPerUtxoByte", 0), - cost_models=result.get("costModels", {}), - ) + if not self._protocol_param or self._is_chain_tip_updated(): + self._protocol_param = self._fetch_protocol_param() + return self._protocol_param - 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") + def _fetch_protocol_param(self) -> ProtocolParameters: + result = self._query_current_protocol_params() + param = ProtocolParameters( + min_fee_constant=result["minFeeConstant"], + min_fee_coefficient=result["minFeeCoefficient"], + max_block_size=result["maxBlockBodySize"], + max_tx_size=result["maxTxSize"], + max_block_header_size=result["maxBlockHeaderSize"], + key_deposit=result["stakeKeyDeposit"], + pool_deposit=result["poolDeposit"], + pool_influence=self._fraction_parser(result["poolInfluence"]), + monetary_expansion=self._fraction_parser(result["monetaryExpansion"]), + treasury_expansion=self._fraction_parser(result["treasuryExpansion"]), + decentralization_param=self._fraction_parser( + result.get("decentralizationParameter", "0/1") + ), + extra_entropy=result.get("extraEntropy", ""), + protocol_major_version=result["protocolVersion"]["major"], + protocol_minor_version=result["protocolVersion"]["minor"], + min_pool_cost=result["minPoolCost"], + price_mem=self._fraction_parser(result["prices"]["memory"]), + price_step=self._fraction_parser(result["prices"]["steps"]), + max_tx_ex_mem=result["maxExecutionUnitsPerTransaction"]["memory"], + max_tx_ex_steps=result["maxExecutionUnitsPerTransaction"]["steps"], + max_block_ex_mem=result["maxExecutionUnitsPerBlock"]["memory"], + max_block_ex_steps=result["maxExecutionUnitsPerBlock"]["steps"], + max_val_size=result["maxValueSize"], + collateral_percent=result["collateralPercentage"], + max_collateral_inputs=result["maxCollateralInputs"], + coins_per_utxo_word=result.get( + "coinsPerUtxoWord", ALONZO_COINS_PER_UTXO_WORD + ), + coins_per_utxo_byte=result.get("coinsPerUtxoByte", 0), + cost_models=result.get("costModels", {}), + ) - args = {"query": "genesisConfig"} - result = self._request(OgmiosQueryType.Query, args) - param.min_utxo = result["protocolParameters"]["minUtxoValue"] + 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") - self._protocol_param = param - return self._protocol_param + result = self._query_genesis_config() + param.min_utxo = result["protocolParameters"]["minUtxoValue"] + return param @property def genesis_param(self) -> GenesisParameters: """Get chain genesis parameters""" - args = {"query": "genesisConfig"} - if not self._genesis_param or self._check_chain_tip_and_update(): - result = self._request(OgmiosQueryType.Query, args) - system_start_unix = int( - calendar.timegm( - time.strptime( - result["systemStart"].split(".")[0], "%Y-%m-%dT%H:%M:%S" - ), - ) - ) - self._genesis_param = GenesisParameters( - active_slots_coefficient=self._fraction_parser( - result["activeSlotsCoefficient"] - ), - update_quorum=result["updateQuorum"], - max_lovelace_supply=result["maxLovelaceSupply"], - network_magic=result["networkMagic"], - epoch_length=result["epochLength"], - system_start=system_start_unix, - slots_per_kes_period=result["slotsPerKesPeriod"], - slot_length=result["slotLength"], - max_kes_evolutions=result["maxKesEvolutions"], - security_param=result["securityParameter"], - ) + if not self._genesis_param or self._is_chain_tip_updated(): + self._genesis_param = self._fetch_genesis_param() return self._genesis_param + def _fetch_genesis_param(self) -> GenesisParameters: + result = self._query_genesis_config() + system_start_unix = int( + calendar.timegm( + time.strptime(result["systemStart"].split(".")[0], "%Y-%m-%dT%H:%M:%S"), + ) + ) + return GenesisParameters( + active_slots_coefficient=self._fraction_parser( + result["activeSlotsCoefficient"] + ), + update_quorum=result["updateQuorum"], + max_lovelace_supply=result["maxLovelaceSupply"], + network_magic=result["networkMagic"], + epoch_length=result["epochLength"], + system_start=system_start_unix, + slots_per_kes_period=result["slotsPerKesPeriod"], + slot_length=result["slotLength"], + max_kes_evolutions=result["maxKesEvolutions"], + security_param=result["securityParameter"], + ) + @property def network(self) -> Network: """Get current network""" @@ -190,37 +214,29 @@ def network(self) -> Network: @property def epoch(self) -> int: """Current epoch number""" - args = {"query": "currentEpoch"} - return self._request(OgmiosQueryType.Query, args) + return self._query_current_epoch() @property def last_block_slot(self) -> int: """Slot number of last block""" - args = {"query": "chainTip"} - return self._request(OgmiosQueryType.Query, args)["slot"] - - def _extract_asset_info(self, asset_hash: str) -> Tuple[str, ScriptHash, AssetName]: - policy_hex, asset_name_hex = asset_hash.split(".") - policy = ScriptHash.from_primitive(policy_hex) - asset_name = AssetName.from_primitive(asset_name_hex) + result = self._query_chain_tip() + return result["slot"] - return policy_hex, policy, asset_name - - def _check_utxo_unspent(self, tx_id: str, index: int) -> bool: - """Check whether an UTxO is unspent with Ogmios. + def utxos(self, address: str) -> List[UTxO]: + """Get all UTxOs associated with an address. Args: - tx_id (str): transaction id. - index (int): transaction index. - """ - - args = {"query": {"utxo": [{"txId": tx_id, "index": index}]}} - results = self._request(OgmiosQueryType.Query, args) + address (str): An address encoded with bech32. - if results: - return True + Returns: + List[UTxO]: A list of UTxOs. + """ + if self._kupo_url: + utxos = self._utxos_kupo(address) else: - return False + utxos = self._utxos_ogmios(address) + + return utxos def _utxos_kupo(self, address: str) -> List[UTxO]: """Get all UTxOs associated with an address with Kupo. @@ -234,7 +250,9 @@ def _utxos_kupo(self, address: str) -> List[UTxO]: List[UTxO]: A list of UTxOs. """ if self._kupo_url is None: - raise AssertionError("kupo_url object attribute has not been assigned properly.") + raise AssertionError( + "kupo_url object attribute has not been assigned properly." + ) address_url = self._kupo_url + "/" + address results = requests.get(address_url).json() @@ -288,6 +306,23 @@ def _utxos_kupo(self, address: str) -> List[UTxO]: return utxos + 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. + """ + results = self._query_utxos_by_tx_id(tx_id, index) + return len(results) > 0 + + def _extract_asset_info(self, asset_hash: str) -> Tuple[str, ScriptHash, AssetName]: + policy_hex, asset_name_hex = asset_hash.split(".") + policy = ScriptHash.from_primitive(policy_hex) + asset_name = AssetName.from_primitive(asset_name_hex) + + return policy_hex, policy, asset_name + def _utxos_ogmios(self, address: str) -> List[UTxO]: """Get all UTxOs associated with an address with Ogmios. @@ -297,12 +332,9 @@ def _utxos_ogmios(self, address: str) -> List[UTxO]: Returns: List[UTxO]: A list of UTxOs. """ - - args = {"query": {"utxo": [address]}} - results = self._request(OgmiosQueryType.Query, args) + results = self._query_utxos_by_address(address) utxos = [] - for result in results: in_ref = result[0] output = result[1] @@ -360,22 +392,6 @@ def _utxos_ogmios(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. diff --git a/pycardano/hash.py b/pycardano/hash.py index df072905..975da671 100644 --- a/pycardano/hash.py +++ b/pycardano/hash.py @@ -1,6 +1,6 @@ """All type of hashes in Cardano ledger spec.""" -from typing import TypeVar, Union, Type +from typing import Type, TypeVar, Union from pycardano.serialization import CBORSerializable diff --git a/pycardano/metadata.py b/pycardano/metadata.py index d8ceaf70..d11ea642 100644 --- a/pycardano/metadata.py +++ b/pycardano/metadata.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Any, ClassVar, List, Union, Type +from typing import Any, ClassVar, List, Type, Union from cbor2 import CBORTag from nacl.encoding import RawEncoder diff --git a/pycardano/nativescript.py b/pycardano/nativescript.py index 5c9c4353..c71d482c 100644 --- a/pycardano/nativescript.py +++ b/pycardano/nativescript.py @@ -3,7 +3,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import ClassVar, List, Union, Type +from typing import ClassVar, List, Type, Union from nacl.encoding import RawEncoder from nacl.hash import blake2b diff --git a/pycardano/plutus.py b/pycardano/plutus.py index 6a7eae36..179f0936 100644 --- a/pycardano/plutus.py +++ b/pycardano/plutus.py @@ -6,7 +6,7 @@ import json from dataclasses import dataclass, field, fields from enum import Enum -from typing import Any, ClassVar, List, Optional, Union, Type +from typing import Any, ClassVar, List, Optional, Type, Union import cbor2 from cbor2 import CBORTag diff --git a/pycardano/transaction.py b/pycardano/transaction.py index db5cac40..d870aaa0 100644 --- a/pycardano/transaction.py +++ b/pycardano/transaction.py @@ -5,7 +5,7 @@ from copy import deepcopy from dataclasses import dataclass, field from pprint import pformat -from typing import Any, Callable, List, Optional, Union, Type +from typing import Any, Callable, List, Optional, Type, Union import cbor2 from cbor2 import CBORTag @@ -311,7 +311,9 @@ def to_shallow_primitive(self) -> List[Primitive]: return [self._TYPE, data] @classmethod - def from_primitive(cls: Type[_DatumOption], values: List[Primitive]) -> _DatumOption: + def from_primitive( + cls: Type[_DatumOption], values: List[Primitive] + ) -> _DatumOption: if values[0] == 0: return _DatumOption(DatumHash(values[1])) else: @@ -418,7 +420,9 @@ def to_primitive(self) -> Primitive: ).to_primitive() @classmethod - def from_primitive(cls: Type[TransactionOutput], value: Primitive) -> TransactionOutput: + def from_primitive( + cls: Type[TransactionOutput], value: Primitive + ) -> TransactionOutput: if isinstance(value, list): output = _TransactionOutputLegacy.from_primitive(value) return cls(output.address, output.amount, datum=output.datum_hash)