274 lines
8.8 KiB
Python
274 lines
8.8 KiB
Python
"""# Partisia Blockchain Client
|
|
|
|
Unofficial library for reading contract states from Partisia Blockchain.
|
|
|
|
This library is not officially associated with Partisia Blockchain nor Partisia
|
|
Group ApS.
|
|
"""
|
|
|
|
import base64
|
|
import dataclasses
|
|
import datetime
|
|
import email.utils
|
|
import json
|
|
import logging
|
|
from collections.abc import Mapping
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
import pbcabi
|
|
import requests
|
|
from frozendict import frozendict
|
|
|
|
from ._version import __version__ # noqa: F401
|
|
from .pbc_types import (
|
|
Address,
|
|
SignedTransaction,
|
|
)
|
|
from .crypto import sign_transaction, SenderAuthentication
|
|
import time
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
HOSTNAME_MAINNET = 'reader.partisiablockchain.com'
|
|
HOSTNAME_TESTNET = 'node1.testnet.partisiablockchain.com'
|
|
|
|
URL_ACCOUNT_PLUGIN = 'https://{hostname}/{shard}blockchain/accountPlugin/local'
|
|
URL_ACCOUNT_PLUGIN_GLOBAL = 'https://{hostname}/{shard}blockchain/accountPlugin/global'
|
|
URL_TRANSACTION = 'https://{hostname}/{shard}blockchain/transaction'
|
|
URL_CONTRACT_STATE = 'https://{hostname}/{shard}blockchain/contracts/{address}?requireContractState=false'
|
|
URL_CHAIN_ID = 'https://{hostname}/blockchain/chainId'
|
|
URL_NONCE = 'https://{hostname}/{shard}blockchain/account/{address}'
|
|
|
|
MPC_DECIMALS = 10000
|
|
|
|
TRANSACTION_VALIDITY_DURATION = 60
|
|
|
|
|
|
def shard_id_for_address(address: str | Address) -> str:
|
|
address = str(address)
|
|
# Very rough implementation
|
|
if address is None:
|
|
msg = 'Address must not be None'
|
|
raise TypeError(msg)
|
|
if address.endswith(('a','d', 'c')):
|
|
return 'shards/Shard0/'
|
|
if address.endswith('2'):
|
|
return 'shards/Shard1/'
|
|
return 'shards/Shard2/'
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class Balances:
|
|
update_time: datetime.datetime
|
|
mpc: Decimal
|
|
byoc: Mapping[str, Decimal]
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class PbcClient:
|
|
session: requests.Session
|
|
sender_authentication: SenderAuthentication | None = None
|
|
hostname: str = HOSTNAME_MAINNET
|
|
|
|
def __post_init__(self):
|
|
assert isinstance(self.session, requests.Session)
|
|
assert isinstance(self.hostname, str)
|
|
|
|
if self.sender_authentication is not None:
|
|
assert isinstance(self.sender_authentication, SenderAuthentication)
|
|
|
|
def send_transaction(self, contract_address: str, rpc: bytes, gas_cost: int):
|
|
if self.sender_authentication is None:
|
|
msg = 'PbcClient.sender_authentication required for send_transaction'
|
|
raise Exception(msg)
|
|
|
|
valid_to_time: int = int(time.time() + TRANSACTION_VALIDITY_DURATION) * 1000
|
|
signed_transaction = sign_transaction(
|
|
self.sender_authentication,
|
|
self.get_sender_authentication_nonce(),
|
|
valid_to_time,
|
|
gas_cost,
|
|
self.get_chain_id(),
|
|
contract_address,
|
|
rpc,
|
|
)
|
|
|
|
print('RPC', signed_transaction.rpc_serialize())
|
|
print('RPC length', len(signed_transaction.rpc_serialize()))
|
|
|
|
assert len(signed_transaction.inner.transaction.rpc_serialize()) == 25
|
|
assert len(signed_transaction.signature.rpc_serialize()) == 65
|
|
assert len(signed_transaction.inner.rpc_serialize()) == 49
|
|
assert len(signed_transaction.rpc_serialize()) == 114
|
|
|
|
return self.send_signed_transaction(signed_transaction)
|
|
|
|
def send_signed_transaction(self, signed_transaction: SignedTransaction):
|
|
url = URL_TRANSACTION.format(
|
|
hostname=self.hostname,
|
|
shard=shard_id_for_address(
|
|
signed_transaction.inner.transaction.contract_address,
|
|
),
|
|
)
|
|
|
|
print(signed_transaction)
|
|
transaction_payload: str = base64.b64encode(
|
|
signed_transaction.rpc_serialize(),
|
|
).decode('utf8')
|
|
print(transaction_payload)
|
|
self._get_json(
|
|
url,
|
|
data={'transactionPayload': transaction_payload},
|
|
method='PUT',
|
|
)
|
|
|
|
def _get_json(
|
|
self,
|
|
url: str,
|
|
data: Mapping[str, Any] = frozendict(),
|
|
method='POST',
|
|
) -> tuple[Any, datetime.datetime]:
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
}
|
|
|
|
response = self.session.request(
|
|
method,
|
|
url,
|
|
headers=headers,
|
|
data=json.dumps(data),
|
|
)
|
|
response.raise_for_status()
|
|
date_text = response.headers.get('last-modified') or response.headers.get(
|
|
'date',
|
|
)
|
|
date = email.utils.parsedate_to_datetime(date_text)
|
|
json_data = response.json()
|
|
if json_data is None:
|
|
msg = 'No result data for ' + url
|
|
raise Exception(msg)
|
|
return (json_data, date)
|
|
|
|
def determine_coins(self) -> list[dict[str, Any]]:
|
|
data: dict = {'path': []}
|
|
|
|
url: str = URL_ACCOUNT_PLUGIN_GLOBAL.format(
|
|
hostname=self.hostname,
|
|
shard='',
|
|
)
|
|
|
|
json_data, date = self._get_json(url, data=data)
|
|
return json_data['coins']['coins']
|
|
|
|
def get_account_balances(self, address: str) -> Balances:
|
|
coins = self.determine_coins()
|
|
|
|
url = URL_ACCOUNT_PLUGIN.format(
|
|
hostname=self.hostname,
|
|
shard=shard_id_for_address(address),
|
|
)
|
|
|
|
data: dict = {
|
|
'path': [
|
|
{'type': 'field', 'name': 'accounts'},
|
|
{'type': 'avl', 'keyType': 'BLOCKCHAIN_ADDRESS', 'key': address},
|
|
],
|
|
}
|
|
account_data, date = self._get_json(url, data=data)
|
|
|
|
byoc: dict[str, Decimal] = {}
|
|
mpc = Decimal(account_data['mpcTokens']) / MPC_DECIMALS
|
|
|
|
for coin_idx, amount_data in enumerate(account_data['accountCoins']):
|
|
coin_data = coins[coin_idx]
|
|
byoc_balance = Decimal(amount_data['balance'])
|
|
denominator = Decimal(coin_data['conversionRate']['denominator'])
|
|
native_balance = byoc_balance / denominator
|
|
byoc[coin_data['symbol']] = native_balance
|
|
del coin_idx, coin_data
|
|
|
|
return Balances(date, mpc, byoc)
|
|
|
|
def get_chain_id(self) -> str:
|
|
url = URL_CHAIN_ID.format(
|
|
hostname=self.hostname,
|
|
)
|
|
return self._get_json(url, method='GET')[0]['chainId']
|
|
|
|
def get_sender_authentication_nonce(self) -> int:
|
|
return self.get_nonce_for_account(self.sender_authentication.sender_address())
|
|
|
|
def get_nonce_for_account(self, address: str) -> int:
|
|
url = URL_NONCE.format(
|
|
hostname=self.hostname,
|
|
shard=shard_id_for_address(address),
|
|
address=address,
|
|
)
|
|
print(url)
|
|
return self._get_json(url, method='GET')[0]['nonce']
|
|
|
|
def get_contract_state(self, address: str) -> tuple[dict, datetime.datetime]:
|
|
# TODO: Rename to get_contract_state_json
|
|
url = URL_CONTRACT_STATE.format(
|
|
hostname=self.hostname,
|
|
shard=shard_id_for_address(address),
|
|
address=address,
|
|
)
|
|
data: dict = {'path': []}
|
|
return self._get_json(url, data=data)
|
|
|
|
def get_typed_contract_state(self, address: str) -> tuple[dict, datetime.datetime]:
|
|
"""
|
|
Only suitable for non-governance WASM contracts.
|
|
"""
|
|
file_abi = self.get_contract_abi(address)
|
|
state, server_time = self.get_contract_state(address)
|
|
state_bytes = base64.b64decode(state['state']['data'])
|
|
state_deserialized = file_abi.contract.read_state(state_bytes)
|
|
|
|
return state_deserialized, server_time
|
|
|
|
def get_typed_contract_avl_tree(
|
|
self,
|
|
address: str,
|
|
avl_tree_id: pbcabi.model.AvlTreeId,
|
|
) -> tuple[dict, datetime.datetime]:
|
|
file_abi = self.get_contract_abi(address)
|
|
state, server_time = self.get_contract_state(address)
|
|
for avl_tree in state['avlTrees']:
|
|
if avl_tree['key'] == avl_tree_id.avl_tree_id:
|
|
break
|
|
|
|
data = {}
|
|
for key_and_value in avl_tree['value']['avlTree']:
|
|
key_bytes = base64.b64decode(key_and_value['key']['data']['data'])
|
|
value_bytes = base64.b64decode(key_and_value['value']['data'])
|
|
|
|
key = file_abi.contract.read_state(
|
|
key_bytes,
|
|
avl_tree_id.type_spec.type_key,
|
|
)
|
|
value = file_abi.contract.read_state(
|
|
value_bytes,
|
|
avl_tree_id.type_spec.type_value,
|
|
)
|
|
|
|
data[key] = value
|
|
|
|
# TODO: Type
|
|
|
|
return data, server_time
|
|
|
|
def get_contract_abi(self, address: str) -> pbcabi.model.FileAbi:
|
|
url = URL_CONTRACT_STATE.format(
|
|
hostname=self.hostname,
|
|
shard=shard_id_for_address(address),
|
|
address=address,
|
|
)
|
|
meta, _ = self._get_json(url, method='GET')
|
|
abi_bytes = base64.b64decode(meta['abi'])
|
|
return pbcabi.model.FileAbi.read_from(abi_bytes)
|