From baa20f997291ba685d4fb8cc2e857c9079139d03 Mon Sep 17 00:00:00 2001 From: Yurii Momotenko Date: Tue, 5 Dec 2023 23:45:45 +0200 Subject: [PATCH] Update docs and methods for Query API and NFT API --- README.md | 286 +++++- ankr/advanced_apis.py | 259 ++++-- ankr/types.py | 1955 ++++++++++++++++++++++++++++++++++------- temp.py | 133 --- 4 files changed, 2075 insertions(+), 558 deletions(-) delete mode 100644 temp.py diff --git a/README.md b/README.md index d9c124b..8da2b2a 100644 --- a/README.md +++ b/README.md @@ -146,23 +146,23 @@ Token API - [`get_token_holders_count_history`](#gettokenholderscounthistory--gettokenholderscounthistoryraw) - [`get_token_holders_count`](#gettokenholderscount--gettokenholderscountraw) - [`get_token_price`](#gettokenprice--gettokenpriceraw) -- [`getTokenTransfers`](#gettokentransfers--gettokentransfersraw) +- [`get_token_transfers`](#gettokentransfers--gettokentransfersraw) NFT API -- [`getNFTsByOwner`](#getnftsbyowner) -- [`getNFTMetadata`](#getnftmetadata) -- [`getNFTHolders`](#getnftholders) -- [`getNftTransfers`](#getnfttransfers) +- [`get_nfts`](#getnfts--getnftsraw) +- [`get_nft_metadata`](#getnftmetadata--getnftmetadataraw) +- [`get_nft_holders`](#getnftholders--getnftholdersraw) +- [`get_nft_transfers`](#getnfttransfers--getnfttransfersraw) Query API -- [`getLogs`](#getlogs) -- [`getBlocks`](#getblocks) -- [`getTransactionsByHash`](#gettransactionsbyhash) -- [`getTransactionsByAddress`](#gettransactionsbyaddress) -- [`getBlockchainStats`](#getblockchainstats) -- [`getInteractions`](#getinteractions) +- [`get_logs`](#getlogs--getlogsraw) +- [`get_blocks`](#getblocks--getblocksraw) +- [`get_transaction`](#gettransaction--gettransactionraw) +- [`get_transactions_by_address`](#gettransactionsbyaddress--gettransactionsbyaddressraw) +- [`get_blockchain_stats`](#getblockchainstats--getblockchainstatsraw) +- [`get_interactions`](#getinteractions--getinteractionsraw) Note: some methods are available in *_raw format, allowing to get full reply with syncStatus and control pagination by hands. @@ -179,6 +179,9 @@ Note: some methods are available in *_raw format, allowing to get full reply wit ### Early Access API #### `get_token_price_history` / `get_token_price_history_raw` + +Get a list of history of the price for given contract to given timestamp. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetTokenPriceHistoryRequest @@ -198,6 +201,9 @@ print(result) ``` #### `get_account_balance_historical` / `get_account_balance_historical_raw` + +Get the coin and token balances of the wallet at specified block. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetAccountBalanceHistoricalRequest @@ -216,6 +222,9 @@ print(result) ``` #### `get_internal_transactions_by_block_number` / `get_internal_transactions_by_block_number_raw` + +Get a list of internal transactions in the block. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetInternalTransactionsByBlockNumberRequest @@ -234,6 +243,9 @@ for transaction in result: ``` #### `get_internal_transactions_by_parent_hash` / `get_internal_transactions_by_parent_hash_raw` + +Get a list of internal transactions in the transaction. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetInternalTransactionsByParentHashRequest @@ -255,6 +267,8 @@ for transaction in result: #### `explain_token_price` / `explain_token_price_raw` +Get a list of tokens and pool how price for calculated. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, ExplainTokenPriceRequest @@ -275,6 +289,8 @@ print(estimates) #### `get_account_balance` / `get_account_balance_raw` +Get the coin and token balances of a wallet. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import GetAccountBalanceRequest @@ -293,6 +309,8 @@ for balance in result: #### `get_currencies` / `get_currencies_raw` +Get a list of supported currencies for a given blockchain. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetCurrenciesRequest @@ -311,6 +329,8 @@ for currency in result: #### `get_token_holders` / `get_token_holders_raw` +Get the list of token holders for a given contract address. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetTokenHoldersRequest @@ -330,6 +350,8 @@ for balance in result: #### `get_token_holders_count_history` / `get_token_holders_count_history_raw` +Get historical data about the number of token holders for a given contract address. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetTokenHoldersCountRequest @@ -349,6 +371,8 @@ for balance in result: #### `get_token_holders_count` / `get_token_holders_count_raw` +Get current data about the number of token holders for a given contract address. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetTokenHoldersCountRequest @@ -367,6 +391,8 @@ print(result) #### `get_token_price` / `get_token_price_raw` +Get token price by contract. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetTokenPriceRequest @@ -385,6 +411,8 @@ print(result) #### `get_token_transfers` / `get_token_transfers_raw` +Get token transfers of specified address. + ```python3 from ankr import AnkrAdvancedAPI, AnkrWeb3 from ankr.types import Blockchain, GetTransfersRequest @@ -402,7 +430,241 @@ result = AnkrAdvancedAPI.get_token_transfers( ) for transfer in result: - print(result) + print(transfer) +``` + +### NFT API + +#### `get_nfts` / `get_nfts_raw` + +Get data about all the NFTs (collectibles) owned by a wallet. + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import Blockchain, GetNFTsByOwnerRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +result = AnkrAdvancedAPI.get_nfts_raw( + request=GetNFTsByOwnerRequest( + blockchain=Blockchain.Eth, + walletAddress='0x0E11A192d574b342C51be9e306694C41547185DD', + ) +) + +for nft in result: + print(nft) +``` + +#### `get_nft_metadata` / `get_nft_metadata_raw` + +Get NFT's contract metadata. + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import Blockchain, GetNFTMetadataRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +metadata, attributes = AnkrAdvancedAPI.get_nft_metadata( + request=GetNFTMetadataRequest( + blockchain=Blockchain.Eth, + contractAddress='0xbc4ca0eda7647a8ab7c2061c2e118a18a936f13d', + tokenId='1500', + forceFetch=False, + ) +) + +print(metadata) +print(attributes) +``` + +#### `get_nft_holders` / `get_nft_holders_raw` + +Get NFT's holders. + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import Blockchain, GetNFTHoldersRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +result = AnkrAdvancedAPI.get_nft_holders( + request=GetNFTHoldersRequest( + blockchain=Blockchain.Arbitrum, + contractAddress='0xc36442b4a4522e871399cd717abdd847ab11fe88', + ), + limit=1000 +) + +for holder in result: + print(holder) +``` + +#### `get_nft_transfers` / `get_nft_transfers_raw` + +Get NFT Transfers of specified address. + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import Blockchain, GetTransfersRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +result = AnkrAdvancedAPI.get_nft_transfers( + request=GetTransfersRequest( + blockchain=[Blockchain.Eth, Blockchain.Bsc], + address=['0xd8da6bf26964af9d7eed9e03e53415d37aa96045'], + fromTimestamp=1672553107, + toTimestamp=1672683207, + ) +) + +for transfer in result: + print(transfer) +``` + +### Query API + +#### `get_logs` / `get_logs_raw` + +Get logs matching the filter. + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import Blockchain, GetLogsRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +result = AnkrAdvancedAPI.get_logs( + request=GetLogsRequest( + blockchain=[Blockchain.Eth], + fromBlock=1181739, + toBlock=1181739, + address=["0x3589d05a1ec4af9f65b0e5554e645707775ee43c"], + topics=[ + [], + ["0x000000000000000000000000feb92d30bf01ff9a1901666c5573532bfa07eeec"], + ], + decodeLogs=True, + ), + limit=10 +) + +for log in result: + print(log) +``` + +#### `get_blocks` / `get_blocks_raw` + +Query data about blocks within a specified range. + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import Blockchain, GetBlocksRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +result = AnkrAdvancedAPI.get_blocks( + request=GetBlocksRequest( + blockchain=Blockchain.Eth, + fromBlock=14500001, + toBlock=14500004, + descOrder=True, + includeLogs=True, + includeTxs=True, + decodeLogs=True, + ) +) + +for block in result: + print(block) +``` + +#### `get_transaction` / `get_transaction_raw` + +Query data about transaction by the transaction hash. + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import GetTransactionsByHashRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +result = AnkrAdvancedAPI.get_transaction( + request=GetTransactionsByHashRequest( + transactionHash='0x82c13aaac6f0b6471afb94a3a64ae89d45baa3608ad397621dbb0d847f51196f', + decodeTxData=True + ) +) + +print(result) +``` + +#### `get_transactions_by_address` / `get_transactions_by_address_raw` + +Query data about transactions of specified address. + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import Blockchain, GetTransactionsByAddressRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +result = AnkrAdvancedAPI.get_transactions_by_address( + request=GetTransactionsByAddressRequest( + blockchain=Blockchain.Bsc, + fromBlock=23593283, + toBlock=23593283, + address=[ + "0x97242e3315c7ece760dc7f83a7dd8af6659f8c4c" + ], + descOrder=True, + ) +) + +for transaction in result: + print(transaction) +``` + +#### `get_blockchain_stats` / `get_blockchain_stats_raw` + +Returns blockchain stats (num of txs, etc). + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import Blockchain, GetBlockchainStatsRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +result = AnkrAdvancedAPI.get_blockchain_stats_raw( + request=GetBlockchainStatsRequest( + blockchain=Blockchain.Bsc, + ) +) + +for stat in result: + print(stat) +``` + +#### `get_interactions` / `get_interactions_raw` + +Returns on which chain address was interacting. + +```python3 +from ankr import AnkrAdvancedAPI, AnkrWeb3 +from ankr.types import GetInteractionsRequest + +ankr_w3 = AnkrWeb3("YOUR-TOKEN") + +result = AnkrAdvancedAPI.get_interactions( + request=GetInteractionsRequest( + address='0xF977814e90dA44bFA03b6295A0616a897441aceC', + ) +) + +for blockchain in result: + print(blockchain) ``` diff --git a/ankr/advanced_apis.py b/ankr/advanced_apis.py index cfb3cee..bcbfe0f 100644 --- a/ankr/advanced_apis.py +++ b/ankr/advanced_apis.py @@ -125,6 +125,7 @@ def get_internal_transactions_by_parent_hash_raw( return reply + class AnkrQueryAPI(AnkrMultichainAPI): def get_logs( self, @@ -153,57 +154,131 @@ def get_logs_raw( return reply - def get_blocks( self, - request: types_gen.GetBlocksRequest, - ) -> List[types_gen.Block]: + request: types.GetBlocksRequest, + ) -> List[types.Block]: reply = self.provider.call_method( rpc="ankr_getBlocks", request=request, - reply=types_gen.GetBlocksReply, + reply=types.GetBlocksReply, ) return reply.blocks def get_blocks_raw( self, - request: types_gen.GetBlocksRequest, - ) -> types_gen.GetBlocksReply: + request: types.GetBlocksRequest, + ) -> types.GetBlocksReply: reply = self.provider.call_method( rpc="ankr_getBlocks", request=request, - reply=types_gen.GetBlocksReply, + reply=types.GetBlocksReply, ) return reply - def get_transaction( self, - transaction_hash: str, - blockchain: Optional[types.BlockchainNames] = None, - include_logs: Optional[bool] = False, - decode_logs: Optional[bool] = False, - decode_tx_data: Optional[bool] = False, - **kwargs: Any, - ) -> Optional[Any]: - reply = super().call_method( - "ankr_getTransactionsByHash", - request_params = ankr_gen.AnkrGetTransactionsByHashParams( - blockchain=blockchain, - transaction_hash=transaction_hash, - include_logs=include_logs, - decode_logs=decode_logs, - decode_tx_data=decode_tx_data, - **kwargs, - ), - call_function=self.api.ankr_get_transactions_by_hash_post, - body_type=ankr_gen.AnkrGetTransactionsByHashBody - ) - - if reply['transactions']: - return reply['transactions'][0] + request: types.GetTransactionsByHashRequest, + ) -> types.Transaction | None: + reply = self.provider.call_method( + rpc="ankr_getTransactionsByHash", + request=request, + reply=types.GetTransactionsByHashReply, + ) + + if len(reply.transactions) > 0: + return reply.transactions[0] + else: + return None + + def get_transaction_raw( + self, + request: types.GetTransactionsByHashRequest, + ) -> types.GetTransactionsByHashReply: + reply = self.provider.call_method( + rpc="ankr_getTransactionsByHash", + request=request, + reply=types.GetTransactionsByHashReply, + ) + + return reply + + def get_transactions_by_address( + self, + request: types.GetTransactionsByAddressRequest, + limit: Optional[int] = None, + ) -> Iterable[types.Transaction]: + for transaction in self.provider.call_method_paginated( + rpc="ankr_getTransactionsByAddress", + request=request, + reply=types.GetTransactionsByAddressReply, + iterable_name="transactions", + iterable_type=types.Transaction, + limit=limit + ): + yield transaction + + def get_transactions_by_address_raw( + self, + request: types.GetTransactionsByAddressRequest, + ) -> types.GetTransactionsByAddressReply: + reply = self.provider.call_method( + rpc="ankr_getTransactionsByAddress", + request=request, + reply=types.GetTransactionsByAddressReply, + ) + + return reply + + def get_blockchain_stats( + self, + request: types.GetBlockchainStatsRequest, + ) -> List[types.BlockchainStats]: + reply = self.provider.call_method( + rpc="ankr_getBlockchainStats", + request=request, + reply=types.GetBlockchainStatsReply, + ) + + return reply.stats + + def get_blockchain_stats_raw( + self, + request: types.GetBlockchainStatsRequest, + ) -> types.GetBlockchainStatsReply: + reply = self.provider.call_method( + rpc="ankr_getBlockchainStats", + request=request, + reply=types.GetBlockchainStatsReply, + ) + + return reply + + def get_interactions( + self, + request: types.GetInteractionsRequest, + ) -> List[types.Blockchain]: + reply = self.provider.call_method( + rpc="ankr_getInteractions", + request=request, + reply=types.GetInteractionsReply, + ) + + return reply.blockchains + + def get_interactions_raw( + self, + request: types.GetInteractionsRequest, + ) -> types.GetInteractionsReply: + reply = self.provider.call_method( + rpc="ankr_getInteractions", + request=request, + reply=types.GetInteractionsReply, + ) + + return reply class AnkrTokenAPI(AnkrMultichainAPI): @@ -415,74 +490,106 @@ def get_token_transfers_raw( return reply - class AnkrNFTAPI(AnkrMultichainAPI): def get_nfts( - self, - request: types_gen.GetNFTsByOwnerRequest, - limit: Optional[int] = None, - ) -> Iterable[types_gen.GetNFTsByOwnerReply]: + self, + request: types.GetNFTsByOwnerRequest, + limit: Optional[int] = None + ) -> Iterable[types.Nft]: for nft in self.provider.call_method_paginated( rpc="ankr_getNFTsByOwner", request=request, - reply=types_gen.GetNFTsByOwnerReply, + reply=types.GetNFTsByOwnerReply, iterable_name="assets", - iterable_type=types_gen.Nft, + iterable_type=types.Nft, limit=limit ): yield nft def get_nfts_raw( - self, - request: GetNFTsByOwnerRequest, - limit: Optional[int] = None, - ) -> GetNFTsByOwnerReply: + self, + request: types.GetNFTsByOwnerRequest + ) -> types.GetNFTsByOwnerReply: reply = self.provider.call_method( rpc="ankr_getNFTsByOwner", request=request, - reply=GetNFTsByOwnerReply, + reply=types.GetNFTsByOwnerReply, ) - return reply def get_nft_metadata( - self, - blockchain: types.BlockchainName, - contract_address: str, - token_id: str, - **kwargs: Any, - ) -> Any: - return super().call_method( - "ankr_getNFTMetadata", - request_params = ankr_gen.AnkrGetNFTMetadataParams( - blockchain=blockchain, - contract_address=contract_address, - token_id=token_id, - **kwargs, - ), - call_function=self.api.ankr_get_nft_metadata_post, - body_type=ankr_gen.AnkrGetNFTMetadataBody + self, + request: types.GetNFTMetadataRequest + ) -> [types.NftMetadata, types.NftAttributes]: + reply = self.provider.call_method( + rpc="ankr_getNFTMetadata", + request=request, + reply=types.GetNFTMetadataReply, + ) + return reply.metadata, reply.attributes + + def get_nft_metadata_raw( + self, + request: types.GetNFTMetadataRequest + ) -> types.GetNFTMetadataReply: + reply = self.provider.call_method( + rpc="ankr_getNFTMetadata", + request=request, + reply=types.GetNFTMetadataReply, ) + return reply def get_nft_holders( - self, - blockchain: types.BlockchainName, - contract_address: str, - limit: Optional[int] = None, - **kwargs: Any, - ) -> Iterable[types.Address]: - return super().call_method_paginated( - method="ankr_getNFTHolders", - request_params=ankr_gen.AnkrGetNFTHoldersParams( - blockchain=blockchain, - contract_address=contract_address, - **kwargs, - ), + self, + request: types.GetNFTHoldersRequest, + limit: Optional[int] = None, + ) -> Iterable[str]: + for holder in self.provider.call_method_paginated( + rpc="ankr_getNFTHolders", + request=request, + reply=types.GetNFTHoldersReply, iterable_name="holders", - limit=limit, - call_function=self.api.ankr_get_nft_holders_post, - body_type=ankr_gen.AnkrGetNFTHoldersBody + iterable_type=str, + limit=limit + ): + yield holder + + def get_nft_holders_raw( + self, + request: types.GetNFTHoldersRequest, + ) -> types.GetNFTHoldersReply: + reply = self.provider.call_method( + rpc="ankr_getNFTHolders", + request=request, + reply=types.GetNFTHoldersReply, ) + return reply + + def get_nft_transfers( + self, + request: types.GetTransfersRequest, + limit: Optional[int] = None + ) -> Iterable[types.NftTransfer]: + for transfer in self.provider.call_method_paginated( + rpc="ankr_getNftTransfers", + request=request, + reply=types.GetNftTransfersReply, + iterable_name="transfers", + iterable_type=types.NftTransfer, + limit=limit + ): + yield transfer + + def get_nft_transfers_raw( + self, + request: types.GetTransfersRequest + ) -> types.GetNftTransfersReply: + reply = self.provider.call_method( + rpc="ankr_getNftTransfers", + request=request, + reply=types.GetNftTransfersReply, + ) + return reply class AnkrAdvancedAPI(AnkrEarlyAccessAPI, AnkrQueryAPI, AnkrTokenAPI, AnkrNFTAPI): diff --git a/ankr/types.py b/ankr/types.py index f706ae4..157462f 100644 --- a/ankr/types.py +++ b/ankr/types.py @@ -1,408 +1,1689 @@ from __future__ import annotations +from enum import Enum +from typing import Literal, List, Dict + + +class SyncStatus: + def __init__(self, + lag: str, + status: str, + timestamp: float): + self.lag = lag + self.status = status + self.timestamp = timestamp + + +class MethodInput: + def __init__(self, + name: str, + size: float, + type: str, + valueDecoded: str): + self.name = name + self.size = size + self.type = type + self.valueDecoded = valueDecoded + + +class Method: + def __init__(self, + id: str, + inputs: List[MethodInput], + name: str, + signature: str, + string: str, + verified: bool): + self.id = id + self.inputs = inputs + self.name = name + self.signature = signature + self.string = string + self.verified = verified + + +class EventInput: + def __init__(self, + indexed: bool, + name: str, + size: float, + type: str, + valueDecoded: str): + self.indexed = indexed + self.name = name + self.size = size + self.type = type + self.valueDecoded = valueDecoded + + +class Event: + def __init__(self, + anonymous: bool, + id: str, + inputs: List[EventInput], + name: str, + signature: str, + string: str, + verified: bool): + self.anonymous = anonymous + self.id = id + self.inputs = inputs + self.name = name + self.signature = signature + self.string = string + self.verified = verified + + +class Log: + def __init__(self, + address: str, + blockHash: str, + blockNumber: str, + blockchain: Blockchain, + data: str, + logIndex: str, + removed: bool, + topics: List[str], + transactionHash: str, + transactionIndex: str, + event: Event = None): + self.address = address + self.blockHash = blockHash + self.blockNumber = blockNumber + self.blockchain = blockchain + self.data = data + self.logIndex = logIndex + self.removed = removed + self.topics = topics + self.transactionHash = transactionHash + self.transactionIndex = transactionIndex + self.event = event + + +class Transaction: + def __init__(self, + blockHash: str, + blockNumber: str, + from_: str, + transactionIndex: str, + value: str, + gasPrice: str = None, + gas: str = None, + contractAddress: str = None, + cumulativeGasUsed: str = None, + input: str = None, + v: str = None, + r: str = None, + s: str = None, + method: Method = None, + to: str = None, + nonce: str = None, + gasUsed: str = None, + logs: List[Log] = None, + hash: str = None, + status: str = None, + blockchain: str = None, + timestamp: str = None, + type: str = None): + self.blockHash = blockHash + self.blockNumber = blockNumber + self.from_ = from_ + self.transactionIndex = transactionIndex + self.value = value + self.gasPrice = gasPrice + self.gas = gas + self.contractAddress = contractAddress + self.cumulativeGasUsed = cumulativeGasUsed + self.input = input + self.v = v + self.r = r + self.s = s + self.method = method + self.to = to + self.nonce = nonce + self.gasUsed = gasUsed + self.logs = logs + self.hash = hash + self.status = status + self.blockchain = blockchain + self.timestamp = timestamp + self.type = type + + +class Block: + def __init__(self, + difficulty: str, + extraData: str, + gasLimit: str, + gasUsed: str, + hash: str, + logsBloom: str, + miner: str, + mixHash: str, + nonce: str, + number: str, + parentHash: str, + receiptsRoot: str, + sha3Uncles: str, + size: str, + stateRoot: str, + timestamp: str, + totalDifficulty: str, + transactions: List[Transaction], + transactionsRoot: str, + uncles: List[str], + blockchain: str = None): + self.difficulty = difficulty + self.extraData = extraData + self.gasLimit = gasLimit + self.gasUsed = gasUsed + self.hash = hash + self.logsBloom = logsBloom + self.miner = miner + self.mixHash = mixHash + self.nonce = nonce + self.number = number + self.parentHash = parentHash + self.receiptsRoot = receiptsRoot + self.sha3Uncles = sha3Uncles + self.size = size + self.stateRoot = stateRoot + self.timestamp = timestamp + self.totalDifficulty = totalDifficulty + self.transactions = transactions + self.transactionsRoot = transactionsRoot + self.uncles = uncles + self.blockchain = blockchain + + +class GetBlocksReply: + def __init__(self, + blocks: List[Block], + syncStatus: SyncStatus = None): + self.blocks = blocks + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + blocks=data.get('blocks'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetBlocksRequest: + def __init__(self, + blockchain: Blockchain, + fromBlock: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + toBlock: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + descOrder: bool = None, + includeLogs: bool = None, + includeTxs: bool = None, + decodeLogs: bool = None, + decodeTxData: bool = None, + syncCheck: bool = None): + self.blockchain = blockchain + self.fromBlock = fromBlock + self.toBlock = toBlock + self.descOrder = descOrder + self.includeLogs = includeLogs + self.includeTxs = includeTxs + self.decodeLogs = decodeLogs + self.decodeTxData = decodeTxData + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'fromBlock': self.fromBlock, + 'toBlock': self.toBlock, + 'descOrder': self.descOrder, + 'includeLogs': self.includeLogs, + 'includeTxs': self.includeTxs, + 'decodeLogs': self.decodeLogs, + 'decodeTxData': self.decodeTxData, + 'syncCheck': self.syncCheck, + } -import enum -from abc import ABC -from typing import Dict, List, Optional, Type, Union - -import humps -from pydantic import BaseModel - - -class Blockchain(str, enum.Enum): - # mainnet - ETH = "eth" - BSC = "bsc" - POLYGON = "polygon" - FANTOM = "fantom" - ARBITRUM = "arbitrum" - AVALANCHE = "avalanche" - SYSCOIN = "syscoin" - OPTIMISM = "optimism" - POLYGON_ZKEVM = "polygon_zkevm" - ROLLUX = "rollux" - BASE = "base" - FLARE = "flare" - GNOSIS = "gnosis" - SCROLL = "scroll" - LINEA = "linea" - - # testnets - ETH_GOERLI = "eth_goerli" - AVALANCHE_FUJI = "avalanche_fuji" - POLYGON_MUMBAI = "polygon_mumbai" - OPTIMISM_TESTNET = "optimism_testnet" - - # appchain - BAS_METAAPES = "bas_metaapes" - - # appchain testnet - BAS_METAAPES_TESTNET = "bas_metaapes_testnet" - - -class NftContractType(str, enum.Enum): - ERC721 = "ERC721" - ERC1155 = "ERC1155" - UNDEFINED = "UNDEFINED" - - -class BlockNumberName(str, enum.Enum): - latest = "latest" - earliest = "earliest" - - -BlockchainName = Union[Blockchain, str] -BlockchainNames = Union[BlockchainName, List[BlockchainName]] -BlockNumber = Union[int, str, BlockNumberName] -Address = str -AddressOrAddresses = Union[str, List[str]] -Topics = Union[str, List[Union[str, List[str]]]] - - -class RPCModel(BaseModel): - class Config: - alias_generator = humps.camelize - allow_population_by_field_name = True - - -class RPCRequestPaginated(ABC, RPCModel): - page_token: Optional[str] = None - - -class RPCReplyPaginated(ABC, RPCModel): - next_page_token: Optional[str] = None - _iter_type: Type - - -class GetNFTsByOwnerRequest(RPCRequestPaginated): - blockchain: BlockchainNames - wallet_address: str - filter: Optional[List[Dict[str, List[str]]]] = None - page_token: Optional[str] = None - page_size: Optional[int] = None -class Attribute(RPCModel): - trait_type: Optional[str] = None - value: Optional[str] = None - display_type: Optional[str] = None - bunny_id: Optional[str] = None - count: Optional[int] = None - frequency: Optional[str] = None - mp_score: Optional[str] = None - rarity: Optional[str] = None +class GetTransactionsByHashReply: + def __init__(self, + transactions: List[Transaction], + syncStatus: SyncStatus = None): + self.transactions = transactions + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + transactions=data.get('transactions'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetTransactionsByHashRequest: + def __init__(self, + transactionHash: str, + blockchain: Blockchain | List[Blockchain] = None, + includeLogs: bool = None, + decodeLogs: bool = None, + decodeTxData: bool = None, + syncCheck: bool = None): + self.transactionHash = transactionHash + self.blockchain = blockchain + self.includeLogs = includeLogs + self.decodeLogs = decodeLogs + self.decodeTxData = decodeTxData + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'transactionHash': self.transactionHash, + 'blockchain': blockchain_value, + 'includeLogs': self.includeLogs, + 'decodeLogs': self.decodeLogs, + 'decodeTxData': self.decodeTxData, + 'syncCheck': self.syncCheck, + } -class Nft(RPCModel): - blockchain: BlockchainName - name: str - token_id: str - token_url: str - image_url: str - collection_name: str - symbol: str - contract_type: str - contract_address: str - quantity: Optional[str] = None - traits: Optional[List[Attribute]] = None +class GetTransactionsByAddressReply: + def __init__(self, + nextPageToken: str, + transactions: List[Transaction], + syncStatus: SyncStatus = None): + self.nextPageToken = nextPageToken + self.transactions = transactions + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + nextPageToken=data.get('nextPageToken'), + transactions=data.get('transactions'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetTransactionsByAddressRequest: + def __init__(self, + address: List[str], + blockchain: Blockchain | List[Blockchain], + fromBlock: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + toBlock: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + fromTimestamp: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + toTimestamp: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + pageToken: str = None, + pageSize: float = None, + descOrder: bool = None, + includeLogs: bool = None, + syncCheck: bool = None): + self.address = address + self.blockchain = blockchain + self.fromBlock = fromBlock + self.toBlock = toBlock + self.fromTimestamp = fromTimestamp + self.toTimestamp = toTimestamp + self.pageToken = pageToken + self.pageSize = pageSize + self.descOrder = descOrder + self.includeLogs = includeLogs + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'address': self.address, + 'blockchain': blockchain_value, + 'fromBlock': self.fromBlock, + 'toBlock': self.toBlock, + 'fromTimestamp': self.fromTimestamp, + 'toTimestamp': self.toTimestamp, + 'pageToken': self.pageToken, + 'pageSize': self.pageSize, + 'descOrder': self.descOrder, + 'includeLogs': self.includeLogs, + 'syncCheck': self.syncCheck, + } -class GetNFTsByOwnerReply(RPCReplyPaginated): - owner: str - assets: List[Nft] - next_page_token: str -class GetNFTMetadataRequest(RPCModel): - blockchain: BlockchainName - contract_address: str - token_id: str +class GetLogsReply: + def __init__(self, + logs: List[Log], + nextPageToken: str = None, + syncStatus: SyncStatus = None): + self.logs = logs + self.nextPageToken = nextPageToken + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + logs=data.get('logs'), + nextPageToken=data.get('nextPageToken'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetLogsRequest: + def __init__(self, + blockchain: Blockchain | List[Blockchain], + fromBlock: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + toBlock: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + fromTimestamp: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + toTimestamp: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + address: List[str] = None, + topics: List[str | List[str]] = None, + pageToken: str = None, + pageSize: float = None, + descOrder: bool = None, + decodeLogs: bool = None, + syncCheck: bool = None): + self.blockchain = blockchain + self.fromBlock = fromBlock + self.toBlock = toBlock + self.fromTimestamp = fromTimestamp + self.toTimestamp = toTimestamp + self.address = address + self.topics = topics + self.pageToken = pageToken + self.pageSize = pageSize + self.descOrder = descOrder + self.decodeLogs = decodeLogs + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'fromBlock': self.fromBlock, + 'toBlock': self.toBlock, + 'fromTimestamp': self.fromTimestamp, + 'toTimestamp': self.toTimestamp, + 'address': self.address, + 'topics': self.topics, + 'pageToken': self.pageToken, + 'pageSize': self.pageSize, + 'descOrder': self.descOrder, + 'decodeLogs': self.decodeLogs, + 'syncCheck': self.syncCheck, + } -class NftAttributes(RPCModel): - token_url: str - image_url: str - name: str - description: str - contract_type: NftContractType - traits: Optional[List[Attribute]] = None +class BlockchainStats: + def __init__(self, + blockTimeMs: float, + blockchain: str, + latestBlockNumber: float, + nativeCoinUsdPrice: str, + totalEventsCount: float, + totalTransactionsCount: float): + self.blockTimeMs = blockTimeMs + self.blockchain = blockchain + self.latestBlockNumber = latestBlockNumber + self.nativeCoinUsdPrice = nativeCoinUsdPrice + self.totalEventsCount = totalEventsCount + self.totalTransactionsCount = totalTransactionsCount + + +class GetBlockchainStatsReply: + def __init__(self, + stats: List[BlockchainStats], + syncStatus: SyncStatus = None): + self.stats = stats + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + stats=data.get('stats'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetBlockchainStatsRequest: + def __init__(self, + blockchain: Blockchain | List[Blockchain] = None, + syncCheck: bool = None): + self.blockchain = blockchain + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'syncCheck': self.syncCheck, + } -class NftMetadata(RPCModel): - blockchain: BlockchainName - contract_address: str - token_id: str - contract_type: NftContractType -class GetNFTMetadataReply(RPCModel): - metadata: Optional[NftMetadata] = None - attributes: Optional[NftAttributes] = None +class GetInteractionsReply: + def __init__(self, + blockchains: List[str], + syncStatus: SyncStatus = None): + self.blockchains = blockchains + self.syncStatus = syncStatus + @classmethod + def from_dict(cls, **data): + return cls( + blockchains=data.get('blockchains'), + syncStatus=data.get('syncStatus'), + ) -class GetNFTHoldersRequest(RPCRequestPaginated): - blockchain: BlockchainName - contract_address: Address - page_token: Optional[str] = None - page_size: Optional[int] = None -class GetNFTHoldersReply(RPCReplyPaginated): - holders: List[Address] - next_page_token: Optional[str] = None +class GetInteractionsRequest: + def __init__(self, + address: str, + syncCheck: bool = None): + self.address = address + self.syncCheck = syncCheck + def to_dict(self): + return { + 'address': self.address, + 'syncCheck': self.syncCheck, + } -class Balance(RPCModel): - blockchain: str - token_name: str - token_symbol: str - token_decimals: int - token_type: str - holder_address: str - balance: str - balance_raw_integer: str - balance_usd: str - token_price: str - thumbnail: str - contract_address: Optional[str] = None -class GetAccountBalanceReply(RPCReplyPaginated): - total_balance_usd: str - assets: List[Balance] - next_page_token: Optional[str] = None +class Balance: + def __init__(self, + balance: str, + balanceRawInteger: str, + balanceUsd: str, + blockchain: Blockchain, + holderAddress: str, + thumbnail: str, + tokenDecimals: float, + tokenName: str, + tokenPrice: str, + tokenSymbol: str, + tokenType: str, + contractAddress: str = None): + self.balance = balance + self.balanceRawInteger = balanceRawInteger + self.balanceUsd = balanceUsd + self.blockchain = blockchain + self.holderAddress = holderAddress + self.thumbnail = thumbnail + self.tokenDecimals = tokenDecimals + self.tokenName = tokenName + self.tokenPrice = tokenPrice + self.tokenSymbol = tokenSymbol + self.tokenType = tokenType + self.contractAddress = contractAddress + + +class GetAccountBalanceReply: + def __init__(self, + assets: List[Balance], + totalBalanceUsd: str, + totalCount: float, + nextPageToken: str = None, + syncStatus: SyncStatus = None): + self.assets = assets + self.totalBalanceUsd = totalBalanceUsd + self.totalCount = totalCount + self.nextPageToken = nextPageToken + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + assets=data.get('assets'), + totalBalanceUsd=data.get('totalBalanceUsd'), + totalCount=data.get('totalCount'), + nextPageToken=data.get('nextPageToken'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetAccountBalanceRequest: + def __init__(self, + walletAddress: str, + blockchain: Blockchain | List[Blockchain] = None, + onlyWhitelisted: bool = None, + nativeFirst: bool = None, + pageToken: str = None, + pageSize: float = None, + syncCheck: bool = None): + self.walletAddress = walletAddress + self.blockchain = blockchain + self.onlyWhitelisted = onlyWhitelisted + self.nativeFirst = nativeFirst + self.pageToken = pageToken + self.pageSize = pageSize + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'walletAddress': self.walletAddress, + 'blockchain': blockchain_value, + 'onlyWhitelisted': self.onlyWhitelisted, + 'nativeFirst': self.nativeFirst, + 'pageToken': self.pageToken, + 'pageSize': self.pageSize, + 'syncCheck': self.syncCheck, + } -class GetAccountBalanceRequest(RPCRequestPaginated): - blockchain: Optional[BlockchainNames] - wallet_address: str - page_token: Optional[str] = None - page_size: Optional[int] = None +class GetTokenPriceReply: + def __init__(self, + blockchain: Blockchain, + usdPrice: str, + contractAddress: str = None, + syncStatus: SyncStatus = None): + self.blockchain = blockchain + self.usdPrice = usdPrice + self.contractAddress = contractAddress + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + blockchain=data.get('blockchain'), + usdPrice=data.get('usdPrice'), + contractAddress=data.get('contractAddress'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetTokenPriceRequest: + def __init__(self, + blockchain: Blockchain, + contractAddress: str = None, + syncCheck: bool = None): + self.blockchain = blockchain + self.contractAddress = contractAddress + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'contractAddress': self.contractAddress, + 'syncCheck': self.syncCheck, + } -class GetTokenHoldersRequest(RPCRequestPaginated): - blockchain: BlockchainName - contract_address: str - page_token: Optional[str] = None - page_size: Optional[int] = None -class HolderBalance(RPCModel): - holder_address: str - balance: str - balance_raw_integer: str +class HolderBalance: + def __init__(self, + balance: str, + balanceRawInteger: str, + holderAddress: str): + self.balance = balance + self.balanceRawInteger = balanceRawInteger + self.holderAddress = holderAddress + + +class GetTokenHoldersReply: + def __init__(self, + blockchain: Blockchain, + contractAddress: str, + holders: List[HolderBalance], + holdersCount: float, + nextPageToken: str, + tokenDecimals: float, + syncStatus: SyncStatus = None): + self.blockchain = blockchain + self.contractAddress = contractAddress + self.holders = holders + self.holdersCount = holdersCount + self.nextPageToken = nextPageToken + self.tokenDecimals = tokenDecimals + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + blockchain=data.get('blockchain'), + contractAddress=data.get('contractAddress'), + holders=data.get('holders'), + holdersCount=data.get('holdersCount'), + nextPageToken=data.get('nextPageToken'), + tokenDecimals=data.get('tokenDecimals'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetTokenHoldersRequest: + def __init__(self, + blockchain: Blockchain, + contractAddress: str, + pageToken: str = None, + pageSize: float = None, + syncCheck: bool = None): + self.blockchain = blockchain + self.contractAddress = contractAddress + self.pageToken = pageToken + self.pageSize = pageSize + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'contractAddress': self.contractAddress, + 'pageToken': self.pageToken, + 'pageSize': self.pageSize, + 'syncCheck': self.syncCheck, + } -class GetTokenHoldersReply(RPCReplyPaginated): - blockchain: BlockchainName - contract_address: str - token_decimals: int - holders: Optional[List[HolderBalance]] - holders_count: int - next_page_token: str +class DailyHolderCount: + def __init__(self, + holderCount: float, + lastUpdatedAt: str, + totalAmount: str, + totalAmountRawInteger: str): + self.holderCount = holderCount + self.lastUpdatedAt = lastUpdatedAt + self.totalAmount = totalAmount + self.totalAmountRawInteger = totalAmountRawInteger + + +class GetTokenHoldersCountReply: + def __init__(self, + blockchain: Blockchain, + contractAddress: str, + holderCountHistory: List[DailyHolderCount], + latestHoldersCount: float, + nextPageToken: str, + tokenDecimals: float, + syncStatus: SyncStatus = None): + self.blockchain = blockchain + self.contractAddress = contractAddress + self.holderCountHistory = holderCountHistory + self.latestHoldersCount = latestHoldersCount + self.nextPageToken = nextPageToken + self.tokenDecimals = tokenDecimals + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + blockchain=data.get('blockchain'), + contractAddress=data.get('contractAddress'), + holderCountHistory=data.get('holderCountHistory'), + latestHoldersCount=data.get('latestHoldersCount'), + nextPageToken=data.get('nextPageToken'), + tokenDecimals=data.get('tokenDecimals'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetTokenHoldersCountRequest: + def __init__(self, + blockchain: Blockchain, + contractAddress: str, + pageToken: str = None, + pageSize: float = None, + syncCheck: bool = None): + self.blockchain = blockchain + self.contractAddress = contractAddress + self.pageToken = pageToken + self.pageSize = pageSize + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'contractAddress': self.contractAddress, + 'pageToken': self.pageToken, + 'pageSize': self.pageSize, + 'syncCheck': self.syncCheck, + } -class GetTokenHoldersCountRequest(RPCRequestPaginated): - blockchain: BlockchainName - contract_address: str - page_token: Optional[str] = None - page_size: Optional[int] = None -class GetTokenPriceRequest(RPCModel): - blockchain: BlockchainName - contract_address: str +class CurrencyDetailsExtended: + def __init__(self, + blockchain: Blockchain, + decimals: float, + name: str, + symbol: str, + thumbnail: str, + address: str = None): + self.blockchain = blockchain + self.decimals = decimals + self.name = name + self.symbol = symbol + self.thumbnail = thumbnail + self.address = address + + +class GetCurrenciesReply: + def __init__(self, + currencies: List[CurrencyDetailsExtended], + syncStatus: SyncStatus = None): + self.currencies = currencies + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + currencies=data.get('currencies'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetCurrenciesRequest: + def __init__(self, + blockchain: Blockchain, + syncCheck: bool = None): + self.blockchain = blockchain + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'syncCheck': self.syncCheck, + } -class GetTokenPriceReply(RPCModel): - usd_price: str - blockchain: BlockchainName - contract_address: str +class TokenTransfer: + def __init__(self, + blockHeight: float, + blockchain: str, + thumbnail: str, + timestamp: float, + tokenDecimals: float, + tokenName: str, + tokenSymbol: str, + transactionHash: str, + value: str, + valueRawInteger: str, + fromAddress: str = None, + contractAddress: str = None, + toAddress: str = None, + direction: str = None): + self.blockHeight = blockHeight + self.blockchain = blockchain + self.thumbnail = thumbnail + self.timestamp = timestamp + self.tokenDecimals = tokenDecimals + self.tokenName = tokenName + self.tokenSymbol = tokenSymbol + self.transactionHash = transactionHash + self.value = value + self.valueRawInteger = valueRawInteger + self.fromAddress = fromAddress + self.contractAddress = contractAddress + self.toAddress = toAddress + self.direction = direction + + +class GetTokenTransfersReply: + def __init__(self, + transfers: List[TokenTransfer], + nextPageToken: str = None, + syncStatus: SyncStatus = None): + self.transfers = transfers + self.nextPageToken = nextPageToken + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + transfers=data.get('transfers'), + nextPageToken=data.get('nextPageToken'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetTransfersRequest: + def __init__(self, + blockchain: Blockchain | List[Blockchain], + fromBlock: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + toBlock: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + fromTimestamp: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + toTimestamp: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + address: List[str] = None, + pageToken: str = None, + pageSize: float = None, + descOrder: bool = None, + syncCheck: bool = None): + self.blockchain = blockchain + self.fromBlock = fromBlock + self.toBlock = toBlock + self.fromTimestamp = fromTimestamp + self.toTimestamp = toTimestamp + self.address = address + self.pageToken = pageToken + self.pageSize = pageSize + self.descOrder = descOrder + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'fromBlock': self.fromBlock, + 'toBlock': self.toBlock, + 'fromTimestamp': self.fromTimestamp, + 'toTimestamp': self.toTimestamp, + 'address': self.address, + 'pageToken': self.pageToken, + 'pageSize': self.pageSize, + 'descOrder': self.descOrder, + 'syncCheck': self.syncCheck, + } -class DailyHolderCount(RPCModel): - holder_count: int - total_amount: str - total_amount_raw_integer: str - last_updated_at: str -class GetTokenHoldersCountReply(RPCReplyPaginated): - blockchain: BlockchainName - contract_address: str - token_decimals: int - holder_count_history: List[DailyHolderCount] - next_page_token: str +class Trait: + def __init__(self, + trait_type: str, + value: str): + self.trait_type = trait_type + self.value = value + + +class Nft: + def __init__(self, + blockchain: Blockchain, + collectionName: str, + contractAddress: str, + contractType: Literal[Literal['ERC721'], Literal['ERC1155'], Literal['UNDEFINED']], + imageUrl: str, + name: str, + symbol: str, + tokenId: str, + tokenUrl: str, + quantity: str = None, + traits: List[Trait] = None): + self.blockchain = blockchain + self.collectionName = collectionName + self.contractAddress = contractAddress + self.contractType = contractType + self.imageUrl = imageUrl + self.name = name + self.symbol = symbol + self.tokenId = tokenId + self.tokenUrl = tokenUrl + self.quantity = quantity + self.traits = traits + + +class GetNFTsByOwnerReply: + def __init__(self, + assets: List[Nft], + nextPageToken: str, + owner: str, + syncStatus: SyncStatus = None): + self.assets = assets + self.nextPageToken = nextPageToken + self.owner = owner + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + assets=data.get('assets'), + nextPageToken=data.get('nextPageToken'), + owner=data.get('owner'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetNFTsByOwnerRequest: + def __init__(self, + walletAddress: str, + blockchain: Blockchain | List[Blockchain] = None, + filter: List[Dict[str, List[str]]] = None, + pageToken: str = None, + pageSize: float = None, + syncCheck: bool = None): + self.walletAddress = walletAddress + self.blockchain = blockchain + self.filter = filter + self.pageToken = pageToken + self.pageSize = pageSize + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'walletAddress': self.walletAddress, + 'blockchain': blockchain_value, + 'filter': self.filter, + 'pageToken': self.pageToken, + 'pageSize': self.pageSize, + 'syncCheck': self.syncCheck, + } -class GetCurrenciesRequest(RPCModel): - blockchain: BlockchainName +class NftAttributes: + def __init__(self, + contractType: Literal[Literal['ERC721'], Literal['ERC1155'], Literal['UNDEFINED']], + description: str, + imageUrl: str, + name: str, + tokenUrl: str, + traits: List[Trait] = None): + self.contractType = contractType + self.description = description + self.imageUrl = imageUrl + self.name = name + self.tokenUrl = tokenUrl + self.traits = traits + + +class NftMetadata: + def __init__(self, + blockchain: Blockchain, + collectionName: str, + collectionSymbol: str, + contractAddress: str, + contractType: Literal[Literal['ERC721'], Literal['ERC1155'], Literal['UNDEFINED']], + tokenId: str): + self.blockchain = blockchain + self.collectionName = collectionName + self.collectionSymbol = collectionSymbol + self.contractAddress = contractAddress + self.contractType = contractType + self.tokenId = tokenId + + +class GetNFTMetadataReply: + def __init__(self, + metadata: NftMetadata = None, + attributes: NftAttributes = None, + syncStatus: SyncStatus = None): + self.metadata = metadata + self.attributes = attributes + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + metadata=data.get('metadata'), + attributes=data.get('attributes'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetNFTMetadataRequest: + def __init__(self, + blockchain: Blockchain, + contractAddress: str, + forceFetch: bool, + tokenId: str, + syncCheck: bool = None): + self.blockchain = blockchain + self.contractAddress = contractAddress + self.forceFetch = forceFetch + self.tokenId = tokenId + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'contractAddress': self.contractAddress, + 'forceFetch': self.forceFetch, + 'tokenId': self.tokenId, + 'syncCheck': self.syncCheck, + } -class CurrencyDetailsExtended(RPCModel): - blockchain: BlockchainName - address: Optional[str] - name: str - decimals: int - symbol: str - thumbnail: str -class GetCurrenciesReply(RPCModel): - currencies: List[CurrencyDetailsExtended] +class GetNFTHoldersReply: + def __init__(self, + holders: List[str], + nextPageToken: str, + syncStatus: SyncStatus = None): + self.holders = holders + self.nextPageToken = nextPageToken + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + holders=data.get('holders'), + nextPageToken=data.get('nextPageToken'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetNFTHoldersRequest: + def __init__(self, + blockchain: Blockchain, + contractAddress: str, + pageToken: str = None, + pageSize: float = None, + syncCheck: bool = None): + self.blockchain = blockchain + self.contractAddress = contractAddress + self.pageToken = pageToken + self.pageSize = pageSize + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'contractAddress': self.contractAddress, + 'pageToken': self.pageToken, + 'pageSize': self.pageSize, + 'syncCheck': self.syncCheck, + } -class GetUsdPriceRequest(RPCModel): - blockchain: BlockchainName - contract_address: str +class NftTransfer: + def __init__(self, + blockHeight: float, + blockchain: Blockchain, + collectionName: str, + collectionSymbol: str, + fromAddress: str, + imageUrl: str, + name: str, + timestamp: float, + toAddress: str, + transactionHash: str, + type: Literal[Literal['ERC721'], Literal['ERC1155'], Literal['UNDEFINED']], + value: str, + tokenId: str = None, + contractAddress: str = None): + self.blockHeight = blockHeight + self.blockchain = blockchain + self.collectionName = collectionName + self.collectionSymbol = collectionSymbol + self.fromAddress = fromAddress + self.imageUrl = imageUrl + self.name = name + self.timestamp = timestamp + self.toAddress = toAddress + self.transactionHash = transactionHash + self.type = type + self.value = value + self.tokenId = tokenId + self.contractAddress = contractAddress + + +class GetNftTransfersReply: + def __init__(self, + transfers: List[NftTransfer], + nextPageToken: str = None, + syncStatus: SyncStatus = None): + self.transfers = transfers + self.nextPageToken = nextPageToken + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + transfers=data.get('transfers'), + nextPageToken=data.get('nextPageToken'), + syncStatus=data.get('syncStatus'), + ) + + + +class GetTokenAllowancesRequest: + def __init__(self, + blockchain: Blockchain | List[Blockchain], + walletAddress: str, + spenderAddress: str = None, + contractAddress: str = None): + self.blockchain = blockchain + self.walletAddress = walletAddress + self.spenderAddress = spenderAddress + self.contractAddress = contractAddress + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'walletAddress': self.walletAddress, + 'spenderAddress': self.spenderAddress, + 'contractAddress': self.contractAddress, + } -class GetUsdPriceReply(RPCModel): - usd_price: str - blockchain: BlockchainName - contract_address: Optional[str] = None -class EventInput(RPCModel): - name: str - type: str - indexed: bool - size: int - value_decoded: str +class ERC20TokenAllowance: + def __init__(self, + blockHeight: float, + thumbnail: str, + timestamp: float, + value: str = None, + tokenDecimals: float = None, + walletAddress: str = None, + contractAddress: str = None, + transactionHash: str = None, + blockchain: str = None, + tokenName: str = None, + tokenSymbol: str = None, + spenderAddress: str = None, + rawLog: Log = None): + self.blockHeight = blockHeight + self.thumbnail = thumbnail + self.timestamp = timestamp + self.value = value + self.tokenDecimals = tokenDecimals + self.walletAddress = walletAddress + self.contractAddress = contractAddress + self.transactionHash = transactionHash + self.blockchain = blockchain + self.tokenName = tokenName + self.tokenSymbol = tokenSymbol + self.spenderAddress = spenderAddress + self.rawLog = rawLog + + +class GetTokenAllowancesReply: + def __init__(self, + allowances: List[ERC20TokenAllowance]): + self.allowances = allowances + + @classmethod + def from_dict(cls, **data): + return cls( + allowances=data.get('allowances'), + ) + + + +class GetTokenPriceHistoryRequest: + def __init__(self, + blockchain: Blockchain, + contractAddress: str, + fromTimestamp: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + toTimestamp: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + interval: float = None, + limit: float = None, + syncCheck: bool = None): + self.blockchain = blockchain + self.contractAddress = contractAddress + self.fromTimestamp = fromTimestamp + self.toTimestamp = toTimestamp + self.interval = interval + self.limit = limit + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'contractAddress': self.contractAddress, + 'fromTimestamp': self.fromTimestamp, + 'toTimestamp': self.toTimestamp, + 'interval': self.interval, + 'limit': self.limit, + 'syncCheck': self.syncCheck, + } -class Event(RPCModel): - name: str - inputs: List[EventInput] - anonymous: bool - string: str - signature: str - id: str - verified: bool +class Quote: + def __init__(self, + blockHeight: float, + timestamp: float, + usdPrice: str): + self.blockHeight = blockHeight + self.timestamp = timestamp + self.usdPrice = usdPrice + + +class GetTokenPriceHistoryReply: + def __init__(self, + quotes: List[Quote], + syncStatus: SyncStatus = None): + self.quotes = quotes + self.syncStatus = syncStatus + + @classmethod + def from_dict(cls, **data): + return cls( + quotes=data.get('quotes'), + syncStatus=data.get('syncStatus'), + ) + + + +class ExplainTokenPriceRequest: + def __init__(self, + blockHeight: float | Literal[Literal['latest']] | Literal[Literal['earliest']], + blockchain: Blockchain, + tokenAddress: str): + self.blockHeight = blockHeight + self.blockchain = blockchain + self.tokenAddress = tokenAddress + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockHeight': self.blockHeight, + 'blockchain': blockchain_value, + 'tokenAddress': self.tokenAddress, + } -class Log(RPCModel): - address: str - topics: List[str] - data: str - block_number: str - transaction_hash: str - transaction_index: str - block_hash: str - log_index: str - removed: bool - event: Optional[Event] = None -class GetLogsReply(RPCReplyPaginated): - logs: List[Log] - next_page_token: Optional[str] = None +class PriceEstimate: + def __init__(self, + price: str, + strategy: str): + self.price = price + self.strategy = strategy + + +class ExplainTokenPriceLPDetails: + def __init__(self, + address: str, + lastUpdatedBlock: float, + price: str, + reserve0: str, + reserve1: str, + token0: str, + token1: str): + self.address = address + self.lastUpdatedBlock = lastUpdatedBlock + self.price = price + self.reserve0 = reserve0 + self.reserve1 = reserve1 + self.token0 = token0 + self.token1 = token1 + + +class ExplainTokenPriceTokenDetails: + def __init__(self, + contractAddress: str, + decimals: float, + name: str, + symbol: str): + self.contractAddress = contractAddress + self.decimals = decimals + self.name = name + self.symbol = symbol + + +class ExplainTokenPriceSinglePair: + def __init__(self, + liquidity_pools: List[ExplainTokenPriceLPDetails], + priceEstimates: List[PriceEstimate], + token0: ExplainTokenPriceTokenDetails, + token1: ExplainTokenPriceTokenDetails): + self.liquidity_pools = liquidity_pools + self.priceEstimates = priceEstimates + self.token0 = token0 + self.token1 = token1 + + +class ExplainTokenPriceReply: + def __init__(self, + blockchain: str, + pairs: List[ExplainTokenPriceSinglePair], + priceEstimates: List[PriceEstimate], + tokenAddress: str): + self.blockchain = blockchain + self.pairs = pairs + self.priceEstimates = priceEstimates + self.tokenAddress = tokenAddress + + @classmethod + def from_dict(cls, **data): + return cls( + blockchain=data.get('blockchain'), + pairs=data.get('pairs'), + priceEstimates=data.get('priceEstimates'), + tokenAddress=data.get('tokenAddress'), + ) + + + +class GetInternalTransactionsByParentHashRequest: + def __init__(self, + blockchain: Blockchain, + onlyWithValue: bool, + parentTransactionHash: str, + syncCheck: bool = None): + self.blockchain = blockchain + self.onlyWithValue = onlyWithValue + self.parentTransactionHash = parentTransactionHash + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockchain': blockchain_value, + 'onlyWithValue': self.onlyWithValue, + 'parentTransactionHash': self.parentTransactionHash, + 'syncCheck': self.syncCheck, + } -class GetLogsRequest(RPCRequestPaginated): - blockchain: BlockchainNames - from_block: Optional[BlockNumber] = None - to_block: Optional[BlockNumber] = None - address: Optional[Union[str, List[str]]] = None - topics: Optional[Topics] = None - page_token: Optional[str] = None - page_size: Optional[int] = None - decode_logs: Optional[bool] = None +class GetInternalTransactionsByBlockNumberRequest: + def __init__(self, + blockNumber: float, + blockchain: Blockchain, + onlyWithValue: bool, + syncCheck: bool = None): + self.blockNumber = blockNumber + self.blockchain = blockchain + self.onlyWithValue = onlyWithValue + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'blockNumber': self.blockNumber, + 'blockchain': blockchain_value, + 'onlyWithValue': self.onlyWithValue, + 'syncCheck': self.syncCheck, + } -class GetBlocksRequest(RPCModel): - blockchain: BlockchainName - from_block: Optional[BlockNumber] = None - to_block: Optional[BlockNumber] = None - desc_order: Optional[bool] = None - include_logs: Optional[bool] = None - include_txs: Optional[bool] = None - decode_logs: Optional[bool] = None - decode_tx_data: Optional[bool] = None -class MethodInput(RPCModel): - name: str - type: str - size: int - value_decoded: str +class InternalTransaction: + def __init__(self, + blockHash: str, + blockHeight: float, + blockchain: Blockchain, + callType: str, + fromAddress: str, + gas: float, + gasUsed: float, + input: str, + output: str, + timestamp: str, + toAddress: str, + transactionHash: str, + transactionIndex: float, + value: str, + callPath: str = None, + callStack: List[float] = None, + error: str = None, + contractAddress: str = None): + self.blockHash = blockHash + self.blockHeight = blockHeight + self.blockchain = blockchain + self.callType = callType + self.fromAddress = fromAddress + self.gas = gas + self.gasUsed = gasUsed + self.input = input + self.output = output + self.timestamp = timestamp + self.toAddress = toAddress + self.transactionHash = transactionHash + self.transactionIndex = transactionIndex + self.value = value + self.callPath = callPath + self.callStack = callStack + self.error = error + self.contractAddress = contractAddress + + +class GetInternalTransactionsReply: + def __init__(self, + internalTransactions: List[InternalTransaction], + nextPageToken: str = None): + self.internalTransactions = internalTransactions + self.nextPageToken = nextPageToken + + @classmethod + def from_dict(cls, **data): + return cls( + internalTransactions=data.get('internalTransactions'), + nextPageToken=data.get('nextPageToken'), + ) + + + +class GetAccountBalanceHistoricalRequest: + def __init__(self, + walletAddress: str, + blockchain: Blockchain | List[Blockchain] = None, + onlyWhitelisted: bool = None, + nativeFirst: bool = None, + pageToken: str = None, + pageSize: float = None, + blockHeight: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None, + syncCheck: bool = None): + self.walletAddress = walletAddress + self.blockchain = blockchain + self.onlyWhitelisted = onlyWhitelisted + self.nativeFirst = nativeFirst + self.pageToken = pageToken + self.pageSize = pageSize + self.blockHeight = blockHeight + self.syncCheck = syncCheck + + def to_dict(self): + if isinstance(self.blockchain, str): + blockchain_value = self.blockchain + elif isinstance(self.blockchain, list): + blockchain_value = [block.value if isinstance(block, Blockchain) else block for block in self.blockchain] + elif self.blockchain is not None: + blockchain_value = self.blockchain.value + else: + blockchain_value = None + return { + 'walletAddress': self.walletAddress, + 'blockchain': blockchain_value, + 'onlyWhitelisted': self.onlyWhitelisted, + 'nativeFirst': self.nativeFirst, + 'pageToken': self.pageToken, + 'pageSize': self.pageSize, + 'blockHeight': self.blockHeight, + 'syncCheck': self.syncCheck, + } -class Method(RPCModel): - name: str - inputs: List[MethodInput] - string: str - signature: str - id: str - verified: bool +class GetAccountBalanceHistoricalReply: + def __init__(self, + assets: List[Balance], + totalBalanceUsd: str, + totalCount: float, + nextPageToken: str = None, + syncStatus: SyncStatus = None, + blockHeight: float | Literal[Literal['latest']] | Literal[Literal['earliest']] = None): + self.assets = assets + self.totalBalanceUsd = totalBalanceUsd + self.totalCount = totalCount + self.nextPageToken = nextPageToken + self.syncStatus = syncStatus + self.blockHeight = blockHeight + + @classmethod + def from_dict(cls, **data): + return cls( + assets=data.get('assets'), + totalBalanceUsd=data.get('totalBalanceUsd'), + totalCount=data.get('totalCount'), + nextPageToken=data.get('nextPageToken'), + syncStatus=data.get('syncStatus'), + blockHeight=data.get('blockHeight'), + ) + + +class Blockchain(Enum): + Arbitrum = 'arbitrum' + Avalanche = 'avalanche' + Avalanche_fuji = 'avalanche_fuji' + Base = 'base' + Bsc = 'bsc' + Eth = 'eth' + Eth_goerli = 'eth_goerli' + Fantom = 'fantom' + Flare = 'flare' + Gnosis = 'gnosis' + Linea = 'linea' + Optimism = 'optimism' + Optimism_testnet = 'optimism_testnet' + Polygon = 'polygon' + Polygon_mumbai = 'polygon_mumbai' + Polygon_zkevm = 'polygon_zkevm' + Rollux = 'rollux' + Scroll = 'scroll' + Syscoin = 'syscoin' -class Transaction(RPCModel): - class Config: - fields = { - "from_address": "from", - "to_address": "to", - } - v: str - r: str - s: str - nonce: str - gas: str - gas_price: str - input: str - block_number: str - to_address: Optional[str] - from_address: str - transaction_index: str - block_hash: str - value: str - type: str - contract_address: Optional[str] - cumulative_gas_used: str - gas_used: str - logs: Optional[List[Log]] - hash: str - status: str - blockchain: str - timestamp: str - method: Optional[Method] - - -class Block(RPCModel): - blockchain: str - number: str - hash: str - parent_hash: str - nonce: str - mix_hash: str - sha3_uncles: str - logs_bloom: str - state_root: str - miner: str - difficulty: str - extra_data: str - size: str - gas_limit: str - gas_used: str - timestamp: str - transactions_root: str - receipts_root: str - total_difficulty: str - transactions: List[Transaction] - uncles: List[str] - - -class GetBlocksReply(RPCModel): - blocks: List[Block] - - -class GetTransactionsByHashRequest(RPCModel): - blockchain: Optional[BlockchainNames] - transaction_hash: str - include_logs: Optional[bool] = None - decode_logs: Optional[bool] = None - decode_tx_data: Optional[bool] = None - - -class GetTransactionsByHashReply(RPCModel): - transactions: List[Transaction] diff --git a/temp.py b/temp.py deleted file mode 100644 index dcb8713..0000000 --- a/temp.py +++ /dev/null @@ -1,133 +0,0 @@ -import time - -import ankr -from ankr import types -from ankr.types import Blockchain - -if __name__ == "__main__": - # ankr_w3= ankr.AnkrWeb3(api_key="31000350a6f93afcfdf56edef0b7169e1d87c3ac72d747f74ee4acfa3a3b1c42") - ankr_w3 = ankr.AnkrWeb3(api_key="") - # - # result = ankr_w3.query.get_logs_raw( - # request=types.GetLogsRequest( - # blockchain=[Blockchain.Eth], - # fromBlock=1181739, - # toBlock=1181739, - # address=["0x3589d05a1ec4af9f65b0e5554e645707775ee43c"], - # topics=[ - # [], - # ["0x000000000000000000000000feb92d30bf01ff9a1901666c5573532bfa07eeec"], - # ], - # decodeLogs=True, - # ), - # limit=10 - # ) - # result = ankr_w3.query.get_logs_raw( - # request=types_gen.GetLogsRequest( - # blockchain= types_gen.Blockchain.Eth, - # fromBlock=1181739, - # toBlock=1181739, - # address=["0x3589d05a1ec4af9f65b0e5554e645707775ee43c"], - # topics=[ - # [], - # ["0x000000000000000000000000feb92d30bf01ff9a1901666c5573532bfa07eeec"], - # ], - # decodeLogs= True, - # pageSize=1, - # pageToken="7hWsYQQwY43r58nqZa7ygmKMnmYueRk2JFvH6o5c5cKv2Z2jFYhsgtsgJhHDih8ukNNue4ZYvDYzoT" - # ), - # ) - # - # pairs = ankr_w3.token.explain_token_price_raw( - # request=types.ExplainTokenPriceRequest( - # blockchain=Blockchain.Eth, - # tokenAddress='0x8290333cef9e6d528dd5618fb97a76f268f3edd4', - # blockHeight=17463534, - # ) - # ) - # - # print(pairs.pairs) - - # - # result = ankr_w3.query.get_blocks( - # request=types_gen.GetBlocksRequest( - # blockchain="eth", - # from_block=14500001, - # to_block=14500004, - # desc_order=True, - # include_logs=True, - # include_txs=True, - # decode_logs=True, - # ) - # ) - - # result = ankr_w3.token.get_token_holders( - # blockchain="eth", - # contract_address="0xdac17f958d2ee523a2206206994597c13d831ec7") - - # result = ankr_w3.nft.get_nft_holders( - # blockchain="arbitrum", - # contract_address="0xc36442b4a4522e871399cd717abdd847ab11fe88", - # limit=1000 - # ) - - # result = ankr_w3.token.get_token_holders_count_history( - # blockchain="eth", - # contract_address="0xdac17f958d2ee523a2206206994597c13d831ec7") - - result = ankr_w3.token.get_token_transfers_raw( - request=types.GetTransfersRequest( - blockchain=Blockchain.Eth, - address=['0xf16e9b0d03470827a95cdfd0cb8a8a3b46969b91'], - fromTimestamp=1674441035, - toTimestamp=1674441035, - descOrder=True, - ) - ) - - # for balance in result: - # print(balance) - print(result.transfers) - # result = ankr_w3.nft.get_nfts_raw( - # request=types_gen.GetNFTsByOwnerRequest( - # blockchain=Blockchain.Eth, - # walletAddress="0x0E11A192d574b342C51be9e306694C41547185DD" - # ) - # ) - # - # result = ankr_w3.nft.get_nft_holders( - # blockchain="eth", - # contract_address="0xc36442b4a4522e871399cd717abdd847ab11fe88", - # limit=1000 - # ) - - # result = ankr_w3.query.get_transaction( - # transaction_hash="0x82c13aaac6f0b6471afb94a3a64ae89d45baa3608ad397621dbb0d847f51196f", - # decode_tx_data=True - # ) - - # result = ankr_w3.token.get_token_holders_count( - # blockchain="eth", - # contract_address="0xdac17f958d2ee523a2206206994597c13d831ec7") - - # result = ankr_w3.token.get_token_price( - # blockchain="eth", - # contract_address="") - # - # result = ankr_w3.nft.get_nft_metadata( - # blockchain="eth", - # contract_address="0xbc4ca0eda7647a8ab7c2061c2e118a18a936f13d", - # token_id="1500", - # force_fetch=False - # ) - - # Call the function - # print(len(list(result))) - # print(result) - # print(result.logs) - # print(result.nextPageToken) - # print(result.logs) - # result - # for log in result: - # print(log) - # print() \ No newline at end of file