import urllib.parse
from typing import Any, Callable, Optional, Union, cast
import requests
from multiversx_sdk.core import (Address, Token, TokenComputer, Transaction,
TransactionOnNetwork)
from multiversx_sdk.core.config import LibraryConfig
from multiversx_sdk.core.constants import METACHAIN_ID
from multiversx_sdk.network_providers.account_awaiter import AccountAwaiter
from multiversx_sdk.network_providers.config import NetworkProviderConfig
from multiversx_sdk.network_providers.constants import (
BASE_USER_AGENT, DEFAULT_ACCOUNT_AWAITING_PATIENCE_IN_MILLISECONDS)
from multiversx_sdk.network_providers.errors import (GenericError,
TransactionFetchingError)
from multiversx_sdk.network_providers.http_resources import (
account_from_api_response, account_storage_entry_from_response,
account_storage_from_response, block_from_response,
definition_of_fungible_token_from_api_response,
definition_of_tokens_collection_from_api_response,
smart_contract_query_to_vm_query_request, token_amount_from_api_response,
transaction_cost_estimation_from_response, transaction_from_api_response,
transaction_from_simulate_response,
transactions_from_send_multiple_response,
vm_query_response_to_smart_contract_query_response)
from multiversx_sdk.network_providers.interface import INetworkProvider
from multiversx_sdk.network_providers.proxy_network_provider import \
ProxyNetworkProvider
from multiversx_sdk.network_providers.resources import (
AccountOnNetwork, AccountStorage, AccountStorageEntry, AwaitingOptions,
BlockOnNetwork, FungibleTokenMetadata, GetBlockArguments, NetworkConfig,
NetworkStatus, TokenAmountOnNetwork, TokensCollectionMetadata,
TransactionCostResponse)
from multiversx_sdk.network_providers.shared import convert_tx_hash_to_string
from multiversx_sdk.network_providers.transaction_awaiter import \
TransactionAwaiter
from multiversx_sdk.network_providers.user_agent import extend_user_agent
from multiversx_sdk.smart_contracts.smart_contract_query import (
SmartContractQuery, SmartContractQueryResponse)
[docs]
class ApiNetworkProvider(INetworkProvider):
def __init__(self,
url: str,
address_hrp: Optional[str] = None,
config: Optional[NetworkProviderConfig] = None) -> None:
self.url = url
self.address_hrp = address_hrp or LibraryConfig.default_address_hrp
self.backing_proxy = ProxyNetworkProvider(url, self.address_hrp)
self.config = config if config is not None else NetworkProviderConfig()
self.user_agent_prefix = f"{BASE_USER_AGENT}/api"
extend_user_agent(self.user_agent_prefix, self.config)
[docs]
def get_network_config(self) -> NetworkConfig:
"""Fetches the general configuration of the network."""
return self.backing_proxy.get_network_config()
[docs]
def get_network_status(self, shard: int = METACHAIN_ID) -> NetworkStatus:
"""Fetches the current status of the network."""
return self.backing_proxy.get_network_status(shard)
[docs]
def get_block(self, arguments: GetBlockArguments) -> BlockOnNetwork:
"""Fetches a block by hash."""
if not arguments.block_hash:
raise Exception("Block hash not provided. Please set the `block_hash` in the arguments.")
result = self.do_get_generic(f"blocks/{arguments.block_hash.hex()}")
return block_from_response(result)
[docs]
def get_latest_block(self, shard: Optional[int] = None) -> BlockOnNetwork:
"""Fetches the latest block of a shard."""
result = self.do_get_generic("/blocks/latest")
return block_from_response(result)
[docs]
def get_account(self, address: Address) -> AccountOnNetwork:
"""Fetches account information for a given address."""
response = self.do_get_generic(f'accounts/{address.to_bech32()}')
account = account_from_api_response(response)
return account
[docs]
def get_account_storage(self, address: Address) -> AccountStorage:
"""
Fetches the storage (key-value pairs) of an account.
When decoding the keys, the errors are ignored. Use the raw values if needed.
"""
response: dict[str, Any] = self.do_get_generic(f"address/{address.to_bech32()}/keys")
return account_storage_from_response(response.get("data", {}))
[docs]
def get_account_storage_entry(self, address: Address, entry_key: str) -> AccountStorageEntry:
"""Fetches a specific storage entry of an account."""
key_as_hex = entry_key.encode().hex()
response: dict[str, Any] = self.do_get_generic(f"address/{address.to_bech32()}/key/{key_as_hex}")
return account_storage_entry_from_response(response.get("data", {}), entry_key)
[docs]
def await_account_on_condition(
self, address: Address, condition: Callable[[AccountOnNetwork],
bool],
options: Optional[AwaitingOptions] = None) -> AccountOnNetwork:
"""Waits until an account satisfies a given condition."""
if options is None:
options = AwaitingOptions(patience_in_milliseconds=DEFAULT_ACCOUNT_AWAITING_PATIENCE_IN_MILLISECONDS)
awaiter = AccountAwaiter(
fetcher=self,
polling_interval_in_milliseconds=options.polling_interval_in_milliseconds,
timeout_interval_in_milliseconds=options.timeout_in_milliseconds,
patience_time_in_milliseconds=options.patience_in_milliseconds
)
return awaiter.await_on_condition(address=address, condition=condition)
[docs]
def send_transaction(self, transaction: Transaction) -> bytes:
"""Broadcasts a transaction and returns its hash."""
response = self.do_post_generic("transactions", transaction.to_dictionary())
return bytes.fromhex(response.get('txHash', ''))
[docs]
def simulate_transaction(self, transaction: Transaction, check_signature: bool = False) -> TransactionOnNetwork:
"""Simulates a transaction."""
url = 'transaction/simulate?checkSignature=false'
if check_signature:
url = 'transaction/simulate'
response: dict[str, Any] = self.do_post_generic(url, transaction.to_dictionary())
return transaction_from_simulate_response(transaction, response.get("data", {}).get("result", {}))
[docs]
def estimate_transaction_cost(self, transaction: Transaction) -> TransactionCostResponse:
"""Estimates the cost of a transaction."""
response: dict[str, Any] = self.do_post_generic('transaction/cost', transaction.to_dictionary())
return transaction_cost_estimation_from_response(response.get("data", {}))
[docs]
def send_transactions(self, transactions: list[Transaction]) -> tuple[int, list[bytes]]:
"""
Broadcasts multiple transactions and returns a tuple of (number of accepted transactions, list of transaction hashes).
In the returned list, the order of transaction hashes corresponds to the order of transactions in the input list.
If a transaction is not accepted, its hash is empty in the returned list.
"""
transactions_as_dictionaries = [transaction.to_dictionary() for transaction in transactions]
response: dict[str, Any] = self.do_post_generic('transaction/send-multiple', transactions_as_dictionaries)
return transactions_from_send_multiple_response(response.get("data", {}), len(transactions))
[docs]
def get_transaction(self, transaction_hash: Union[str, bytes]) -> TransactionOnNetwork:
"""Fetches a transaction that was previously broadcasted (maybe already processed by the network)."""
transaction_hash = convert_tx_hash_to_string(transaction_hash)
try:
response = self.do_get_generic(f'transactions/{transaction_hash}')
except GenericError as ge:
raise TransactionFetchingError(ge.url, ge.data)
return transaction_from_api_response(transaction_hash, response)
[docs]
def await_transaction_completed(
self, transaction_hash: Union[str, bytes],
options: Optional[AwaitingOptions] = None) -> TransactionOnNetwork:
"""Waits until the transaction is completely processed."""
transaction_hash = convert_tx_hash_to_string(transaction_hash)
if options is None:
options = AwaitingOptions()
awaiter = TransactionAwaiter(
fetcher=self,
polling_interval_in_milliseconds=options.polling_interval_in_milliseconds,
timeout_interval_in_milliseconds=options.timeout_in_milliseconds,
patience_time_in_milliseconds=options.patience_in_milliseconds
)
return awaiter.await_completed(transaction_hash)
[docs]
def await_transaction_on_condition(
self, transaction_hash: Union[str, bytes],
condition: Callable[[TransactionOnNetwork],
bool],
options: Optional[AwaitingOptions] = None) -> TransactionOnNetwork:
"""Waits until a transaction satisfies a given condition."""
transaction_hash = convert_tx_hash_to_string(transaction_hash)
if options is None:
options = AwaitingOptions()
awaiter = TransactionAwaiter(
fetcher=self,
polling_interval_in_milliseconds=options.polling_interval_in_milliseconds,
timeout_interval_in_milliseconds=options.timeout_in_milliseconds,
patience_time_in_milliseconds=options.patience_in_milliseconds
)
return awaiter.await_on_condition(transaction_hash, condition)
[docs]
def get_token_of_account(self, address: Address, token: Token) -> TokenAmountOnNetwork:
"""
Fetches the balance of an account, for a given token.
Able to handle both fungible and non-fungible tokens (NFTs, SFTs, MetaESDTs).
"""
if token.nonce:
identifier = TokenComputer().compute_extended_identifier(token)
result = self.do_get_generic(f"accounts/{address.to_bech32()}/nfts/{identifier}")
else:
result = self.do_get_generic(f"accounts/{address.to_bech32()}/tokens/{token.identifier}")
return token_amount_from_api_response(result)
[docs]
def get_fungible_tokens_of_account(self, address: Address) -> list[TokenAmountOnNetwork]:
"""
Fetches the balances of an account, for all fungible tokens held by the account.
Pagination isn't explicitly handled by a basic network provider, but can be achieved by using `do_get_generic`.
"""
result: list[dict[str, Any]] = self.do_get_generic(f"accounts/{address.to_bech32()}/tokens")
return [token_amount_from_api_response(token) for token in result]
[docs]
def get_non_fungible_tokens_of_account(self, address: Address) -> list[TokenAmountOnNetwork]:
"""
Fetches the balances of an account, for all non-fungible tokens held by the account.
Pagination isn't explicitly handled by a basic network provider, but can be achieved by using `do_get_generic`.
"""
result: list[dict[str, Any]] = self.do_get_generic(f"accounts/{address.to_bech32()}/nfts")
return [token_amount_from_api_response(token) for token in result]
[docs]
def get_definition_of_fungible_token(self, token_identifier: str) -> FungibleTokenMetadata:
"""Fetches the definition of a fungible token."""
result = self.do_get_generic(f"tokens/{token_identifier}")
return definition_of_fungible_token_from_api_response(result)
[docs]
def get_definition_of_tokens_collection(self, collection_name: str) -> TokensCollectionMetadata:
"""Fetches the definition of a tokens collection."""
result = self.do_get_generic(f"collections/{collection_name}")
return definition_of_tokens_collection_from_api_response(result)
[docs]
def query_contract(self, query: SmartContractQuery) -> SmartContractQueryResponse:
request = smart_contract_query_to_vm_query_request(query)
response = self.do_post_generic('query', request)
return vm_query_response_to_smart_contract_query_response(response, query.function)
[docs]
def do_get_generic(self, url: str, url_parameters: Optional[dict[str, Any]] = None) -> Any:
"""Does a generic GET request against the network(handles API enveloping)."""
url = f'{self.url}/{url}'
if url_parameters is not None:
params = urllib.parse.urlencode(url_parameters)
url = f"{url}?{params}"
response = self._do_get(url)
return response
[docs]
def do_post_generic(
self, url: str, data: Any, url_parameters: Optional[dict[str, Any]] = None) -> Any:
"""Does a generic GET request against the network(handles API enveloping)."""
url = f'{self.url}/{url}'
if url_parameters is not None:
params = urllib.parse.urlencode(url_parameters)
url = f"{url}?{params}"
response = self._do_post(url, data)
return response
def _do_get(self, url: str) -> Any:
try:
response = requests.get(url, **self.config.requests_options)
response.raise_for_status()
parsed = response.json()
return self._get_data(parsed, url)
except requests.HTTPError as err:
error_data = self._extract_error_from_response(err.response)
raise GenericError(url, error_data)
except requests.ConnectionError as err:
raise GenericError(url, err)
except Exception as err:
raise GenericError(url, err)
def _do_post(self, url: str, payload: Any) -> dict[str, Any]:
try:
response = requests.post(url, json=payload, **self.config.requests_options)
response.raise_for_status()
parsed = response.json()
return cast(dict[str, Any], self._get_data(parsed, url))
except requests.HTTPError as err:
error_data = self._extract_error_from_response(err.response)
raise GenericError(url, error_data)
except requests.ConnectionError as err:
raise GenericError(url, err)
except Exception as err:
raise GenericError(url, err)
def _get_data(self, parsed: Any, url: str) -> Any:
if isinstance(parsed, list):
return cast(Any, parsed)
else:
err = parsed.get("error", None)
if err:
code = parsed.get("statusCode")
raise GenericError(url, f"code:{code}, error: {err}")
else:
return parsed
def _extract_error_from_response(self, response: Any):
try:
return response.json()
except Exception:
return response.text