diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 4153a427..3a9975eb 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -32,9 +32,6 @@ jobs: - name: Install dependencies run: | poetry install - - name: Run static analyses - run: | - make qa - name: Run unit tests run: | poetry run pytest --doctest-modules --ignore=examples --cov=pycardano --cov-config=.coveragerc --cov-report=xml @@ -42,6 +39,9 @@ jobs: uses: codecov/codecov-action@v3 with: fail_ci_if_error: true + - name: Run static analyses + run: | + make qa continuous-integration: runs-on: ${{ matrix.os }} diff --git a/pycardano/address.py b/pycardano/address.py index a730138a..cbe07fef 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 +from typing import Union, Type from pycardano.crypto.bech32 import decode, encode from pycardano.exception import ( @@ -160,7 +160,7 @@ def to_primitive(self) -> bytes: return self.encode() @classmethod - def from_primitive(cls, value: bytes) -> PointerAddress: + def from_primitive(cls: Type[PointerAddress], value: bytes) -> PointerAddress: return cls.decode(value) def __eq__(self, other): @@ -339,7 +339,7 @@ def to_primitive(self) -> bytes: return bytes(self) @classmethod - def from_primitive(cls, value: Union[bytes, str]) -> Address: + def from_primitive(cls: Type[Address], value: Union[bytes, str]) -> Address: if isinstance(value, str): value = bytes(decode(value)) header = value[0] diff --git a/pycardano/backend/ogmios.py b/pycardano/backend/ogmios.py index 37c5a906..163a3e6d 100644 --- a/pycardano/backend/ogmios.py +++ b/pycardano/backend/ogmios.py @@ -1,7 +1,8 @@ import calendar import json import time -from typing import Dict, List, Union +from enum import Enum +from typing import Any, Dict, List, Optional, Union, Tuple import cbor2 import requests @@ -32,7 +33,24 @@ __all__ = ["OgmiosChainContext"] +JSON = Dict[str, Any] + + +class OgmiosQueryType(str, Enum): + Query = "Query" + SubmitTx = "SubmitTx" + EvaluateTx = "EvaluateTx" + + class OgmiosChainContext(ChainContext): + _ws_url: str + _network: Network + _service_name: str + _kupo_url: Optional[str] + _last_known_block_slot: int + _genesis_param: Optional[GenesisParameters] + _protocol_param: Optional[ProtocolParameters] + def __init__( self, ws_url: str, @@ -48,7 +66,7 @@ def __init__( self._genesis_param = None self._protocol_param = None - def _request(self, method: str, args: dict) -> Union[dict, int]: + def _request(self, method: OgmiosQueryType, args: JSON) -> Any: ws = websocket.WebSocket() ws.connect(self._ws_url) request = json.dumps( @@ -56,7 +74,7 @@ def _request(self, method: str, args: dict) -> Union[dict, int]: "type": "jsonwsp/request", "version": "1.0", "servicename": self._service_name, - "methodname": method, + "methodname": method.value, "args": args, }, separators=(",", ":"), @@ -86,10 +104,9 @@ def _fraction_parser(fraction: str) -> float: @property def protocol_param(self) -> ProtocolParameters: """Get current protocol parameters""" - method = "Query" args = {"query": "currentProtocolParameters"} if not self._protocol_param or self._check_chain_tip_and_update(): - result = self._request(method, args) + result = self._request(OgmiosQueryType.Query, args) param = ProtocolParameters( min_fee_constant=result["minFeeConstant"], min_fee_coefficient=result["minFeeCoefficient"], @@ -130,7 +147,7 @@ def protocol_param(self) -> ProtocolParameters: param.cost_models["PlutusV2"] = param.cost_models.pop("plutus:v2") args = {"query": "genesisConfig"} - result = self._request(method, args) + result = self._request(OgmiosQueryType.Query, args) param.min_utxo = result["protocolParameters"]["minUtxoValue"] self._protocol_param = param @@ -139,10 +156,9 @@ def protocol_param(self) -> ProtocolParameters: @property def genesis_param(self) -> GenesisParameters: """Get chain genesis parameters""" - method = "Query" args = {"query": "genesisConfig"} if not self._genesis_param or self._check_chain_tip_and_update(): - result = self._request(method, args) + result = self._request(OgmiosQueryType.Query, args) system_start_unix = int( calendar.timegm( time.strptime( @@ -174,23 +190,21 @@ def network(self) -> Network: @property def epoch(self) -> int: """Current epoch number""" - method = "Query" args = {"query": "currentEpoch"} - return self._request(method, args) + return self._request(OgmiosQueryType.Query, args) @property def last_block_slot(self) -> int: """Slot number of last block""" - method = "Query" args = {"query": "chainTip"} - return self._request(method, args)["slot"] + return self._request(OgmiosQueryType.Query, args)["slot"] - def _extract_asset_info(self, asset_hash: str): + 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_hex = AssetName.from_primitive(asset_name_hex) + asset_name = AssetName.from_primitive(asset_name_hex) - return policy_hex, policy, asset_name_hex + 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. @@ -200,9 +214,8 @@ def _check_utxo_unspent(self, tx_id: str, index: int) -> bool: index (int): transaction index. """ - method = "Query" args = {"query": {"utxo": [{"txId": tx_id, "index": index}]}} - results = self._request(method, args) + results = self._request(OgmiosQueryType.Query, args) if results: return True @@ -220,6 +233,9 @@ def _utxos_kupo(self, address: str) -> List[UTxO]: Returns: List[UTxO]: A list of UTxOs. """ + if self._kupo_url is None: + raise AssertionError("kupo_url object attribute has not been assigned properly.") + address_url = self._kupo_url + "/" + address results = requests.get(address_url).json() @@ -282,9 +298,8 @@ def _utxos_ogmios(self, address: str) -> List[UTxO]: List[UTxO]: A list of UTxOs. """ - method = "Query" args = {"query": {"utxo": [address]}} - results = self._request(method, args) + results = self._request(OgmiosQueryType.Query, args) utxos = [] @@ -374,9 +389,8 @@ def submit_tx(self, cbor: Union[bytes, str]): if isinstance(cbor, bytes): cbor = cbor.hex() - method = "SubmitTx" args = {"submit": cbor} - result = self._request(method, args) + result = self._request(OgmiosQueryType.SubmitTx, args) if "SubmitFail" in result: raise TransactionFailedException(result["SubmitFail"]) @@ -395,9 +409,8 @@ def evaluate_tx(self, cbor: Union[bytes, str]) -> Dict[str, ExecutionUnits]: if isinstance(cbor, bytes): cbor = cbor.hex() - method = "EvaluateTx" args = {"evaluate": cbor} - result = self._request(method, args) + result = self._request(OgmiosQueryType.EvaluateTx, args) if "EvaluationResult" not in result: raise TransactionFailedException(result) else: diff --git a/pycardano/hash.py b/pycardano/hash.py index 415a8d77..df072905 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 +from typing import TypeVar, Union, Type from pycardano.serialization import CBORSerializable @@ -67,7 +67,7 @@ def to_primitive(self) -> bytes: return self.payload @classmethod - def from_primitive(cls: T, value: Union[bytes, str]) -> T: + def from_primitive(cls: Type[T], value: Union[bytes, str]) -> T: if isinstance(value, str): value = bytes.fromhex(value) return cls(value) diff --git a/pycardano/key.py b/pycardano/key.py index 87587a7d..23facb37 100644 --- a/pycardano/key.py +++ b/pycardano/key.py @@ -4,6 +4,7 @@ import json import os +from typing import Type from nacl.encoding import RawEncoder from nacl.hash import blake2b @@ -61,7 +62,7 @@ def to_primitive(self) -> bytes: return self.payload @classmethod - def from_primitive(cls, value: bytes) -> Key: + def from_primitive(cls: Type["Key"], value: bytes) -> Key: return cls(value) def to_json(self) -> str: diff --git a/pycardano/metadata.py b/pycardano/metadata.py index 16e5d773..d8ceaf70 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 +from typing import Any, ClassVar, List, Union, Type from cbor2 import CBORTag from nacl.encoding import RawEncoder @@ -91,7 +91,7 @@ def to_primitive(self) -> Primitive: return CBORTag(AlonzoMetadata.TAG, super(AlonzoMetadata, self).to_primitive()) @classmethod - def from_primitive(cls: AlonzoMetadata, value: CBORTag) -> AlonzoMetadata: + def from_primitive(cls: Type[AlonzoMetadata], value: CBORTag) -> AlonzoMetadata: if not hasattr(value, "tag"): raise DeserializeException( f"{value} does not match the data schema of AlonzoMetadata." @@ -111,7 +111,7 @@ def to_primitive(self) -> Primitive: return self.data.to_primitive() @classmethod - def from_primitive(cls: AuxiliaryData, value: Primitive) -> AuxiliaryData: + def from_primitive(cls: Type[AuxiliaryData], value: Primitive) -> AuxiliaryData: for t in [AlonzoMetadata, ShellayMarryMetadata, Metadata]: # The schema of metadata in different eras are mutually exclusive, so we can try deserializing # them one by one without worrying about mismatch. diff --git a/pycardano/nativescript.py b/pycardano/nativescript.py index a615ca80..5c9c4353 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 +from typing import ClassVar, List, Union, Type from nacl.encoding import RawEncoder from nacl.hash import blake2b @@ -27,7 +27,7 @@ class NativeScript(ArrayCBORSerializable): @classmethod def from_primitive( - cls: NativeScript, value: Primitive + cls: Type[NativeScript], value: Primitive ) -> Union[ ScriptPubkey, ScriptAll, ScriptAny, ScriptNofK, InvalidBefore, InvalidHereAfter ]: diff --git a/pycardano/network.py b/pycardano/network.py index dc19c781..86277f59 100644 --- a/pycardano/network.py +++ b/pycardano/network.py @@ -3,6 +3,7 @@ from __future__ import annotations from enum import Enum +from typing import Type from pycardano.serialization import CBORSerializable @@ -21,5 +22,5 @@ def to_primitive(self) -> int: return self.value @classmethod - def from_primitive(cls, value: int) -> Network: + def from_primitive(cls: Type[Network], value: int) -> Network: return cls(value) diff --git a/pycardano/plutus.py b/pycardano/plutus.py index 55bc978d..6a7eae36 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 +from typing import Any, ClassVar, List, Optional, Union, Type import cbor2 from cbor2 import CBORTag @@ -66,7 +66,7 @@ def to_shallow_primitive(self) -> dict: return result @classmethod - def from_primitive(cls: CostModels, value: dict) -> CostModels: + def from_primitive(cls: Type[CostModels], value: dict) -> CostModels: raise DeserializeException( "Deserialization of cost model is impossible, because some information is lost " "during serialization." @@ -480,7 +480,7 @@ def to_shallow_primitive(self) -> CBORTag: return CBORTag(102, [self.CONSTR_ID, primitives]) @classmethod - def from_primitive(cls: PlutusData, value: CBORTag) -> PlutusData: + def from_primitive(cls: Type[PlutusData], value: CBORTag) -> PlutusData: if not isinstance(value, CBORTag): raise DeserializeException( f"Unexpected type: {CBORTag}. Got {type(value)} instead." @@ -643,7 +643,7 @@ def _dfs(obj): return _dfs(self.data) @classmethod - def from_primitive(cls: RawPlutusData, value: CBORTag) -> RawPlutusData: + def from_primitive(cls: Type[RawPlutusData], value: CBORTag) -> RawPlutusData: return cls(value) @@ -675,7 +675,7 @@ def to_primitive(self) -> int: return self.value @classmethod - def from_primitive(cls, value: int) -> RedeemerTag: + def from_primitive(cls: Type[RedeemerTag], value: int) -> RedeemerTag: return cls(value) @@ -704,7 +704,7 @@ class Redeemer(ArrayCBORSerializable): ex_units: ExecutionUnits = None @classmethod - def from_primitive(cls: Redeemer, values: List[Primitive]) -> Redeemer: + def from_primitive(cls: Type[Redeemer], values: List[Primitive]) -> Redeemer: if isinstance(values[2], CBORTag) and cls is Redeemer: values[2] = RawPlutusData.from_primitive(values[2]) redeemer = super(Redeemer, cls).from_primitive( diff --git a/pycardano/serialization.py b/pycardano/serialization.py index f7d0fabb..4385f7e7 100644 --- a/pycardano/serialization.py +++ b/pycardano/serialization.py @@ -222,7 +222,7 @@ def to_validated_primitive(self) -> Primitive: return self.to_primitive() @classmethod - def from_primitive(cls: CBORBase, value: Primitive) -> CBORBase: + def from_primitive(cls: Type[CBORBase], value: Primitive) -> CBORBase: """Turn a CBOR primitive to its original class type. Args: @@ -473,7 +473,7 @@ def to_shallow_primitive(self) -> List[Primitive]: return primitives @classmethod - def from_primitive(cls: ArrayBase, values: List[Primitive]) -> ArrayBase: + def from_primitive(cls: Type[ArrayBase], values: List[Primitive]) -> ArrayBase: """Restore a primitive value to its original class type. Args: @@ -585,7 +585,7 @@ def to_shallow_primitive(self) -> Primitive: return primitives @classmethod - def from_primitive(cls: MapBase, values: Primitive) -> MapBase: + def from_primitive(cls: Type[MapBase], values: Primitive) -> MapBase: """Restore a primitive value to its original class type. Args: @@ -704,7 +704,7 @@ def _get_sortable_val(key): return dict(sorted(self.data.items(), key=lambda x: _get_sortable_val(x[0]))) @classmethod - def from_primitive(cls: DictBase, value: dict) -> DictBase: + def from_primitive(cls: Type[DictBase], value: dict) -> DictBase: """Restore a primitive value to its original class type. Args: diff --git a/pycardano/transaction.py b/pycardano/transaction.py index 0a512365..db5cac40 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 +from typing import Any, Callable, List, Optional, Union, Type import cbor2 from cbor2 import CBORTag @@ -281,7 +281,7 @@ def __post_init__(self): self._TYPE = 2 @classmethod - def from_primitive(cls: _Script, values: List[Primitive]) -> _Script: + def from_primitive(cls: Type[_Script], values: List[Primitive]) -> _Script: if values[0] == 0: return cls(NativeScript.from_primitive(values[1])) elif values[0] == 1: @@ -311,7 +311,7 @@ def to_shallow_primitive(self) -> List[Primitive]: return [self._TYPE, data] @classmethod - def from_primitive(cls: _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: @@ -327,7 +327,7 @@ def to_primitive(self) -> Primitive: return CBORTag(24, cbor2.dumps(self.script, default=default_encoder)) @classmethod - def from_primitive(cls: _ScriptRef, value: Primitive) -> _ScriptRef: + def from_primitive(cls: Type[_ScriptRef], value: Primitive) -> _ScriptRef: return cls(_Script.from_primitive(cbor2.loads(value.value))) @@ -418,7 +418,7 @@ def to_primitive(self) -> Primitive: ).to_primitive() @classmethod - def from_primitive(cls: 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) diff --git a/pyproject.toml b/pyproject.toml index 9272cf7e..a78ff63d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,7 +67,6 @@ python_version = 3.7 exclude = [ '^pycardano/backend/base.py$', '^pycardano/backend/blockfrost.py$', - '^pycardano/backend/ogmios.py$', '^pycardano/cip/cip8.py$', '^pycardano/crypto/bech32.py$', '^pycardano/address.py$',