diff --git a/README.md b/README.md index b312ae7..757e83c 100644 --- a/README.md +++ b/README.md @@ -13,18 +13,27 @@ pip install ankr-sdk #### 2. Initialize the SDK ```python3 -from ankr import AnkrAdvancedAPI, types +from ankr import AnkrWeb3 -ankr_api = AnkrAdvancedAPI() +ankr_w3 = AnkrWeb3() ``` #### 3. Use the sdk and call one of the supported methods +#### Node's API ```python3 -from ankr.types import BlockchainName +eth_block = ankr_w3.eth.get_block("latest") +bsc_block = ankr_w3.bsc.get_block("latest") +polygon_block = ankr_w3.polygon.get_block("latest") +``` + +#### Ankr NFT API: get all addresses' NFTs -nfts = ankr_api.get_nfts( - blockchain=BlockchainName.ETH, +```python3 +from ankr.types import Blockchain + +nfts = ankr_w3.nft.get_nfts( + blockchain=[Blockchain.ETH, Blockchain.BSC], wallet_address="0x0E11A192d574b342C51be9e306694C41547185DD", filter=[ {"0x700b4b9f39bb1faf5d0d16a20488f2733550bff4": []}, @@ -33,7 +42,29 @@ nfts = ankr_api.get_nfts( ) ``` -## Supported chains +#### Ankr Token API: get all wallet's tokens on every supported chain +```python3 +assets = ankr_w3.token.get_account_balance( + wallet_address="0x77A859A53D4de24bBC0CC80dD93Fbe391Df45527" +) +``` + +#### Ankr Query API: search for logs without range limits +```python3 +logs = ankr_w3.query.get_logs( + blockchain="eth", + from_block="0xdaf6b1", + to_block=14350010, + address=["0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"], + topics=[ + [], + ["0x000000000000000000000000def1c0ded9bec7f1a1670819833240f027b25eff"], + ], + decode_logs=True, +) +``` + +## Ankr Advanced APIs supported chains `ankr-sdk` supports the following chains at this time: @@ -49,16 +80,95 @@ nfts = ankr_api.get_nfts( `ankr-sdk` supports the following methods: -- [`get_nfts`](#get_nfts) -- [`get_logs`](#get_logs) -- [`get_blocks`](#get_blocks) +- [`nft.get_nfts`](#get_nfts) +- [`nft.get_nft_metadata`](#get_nft_metadata) +- [`token.get_token_holders`](#get_token_holders) +- [`token.get_token_holders_count_history`](#get_token_holders_count_history) +- [`token.get_token_holders_count`](#get_token_holders_count) +- [`token.get_account_balance`](#get_account_balance) +- [`query.get_logs`](#get_logs) +- [`query.get_blocks`](#get_blocks) +- [`query.get_transaction`](#get_transaction) + +#### `get_nfts` + +Get data about all the NFTs (collectibles) owned by a wallet. + +````python3 +nfts = ankr_w3.nft.get_nfts( + blockchain="eth", + wallet_address="0x0E11A192d574b342C51be9e306694C41547185DD", + filter=[ + {"0x700b4b9f39bb1faf5d0d16a20488f2733550bff4": []}, + {"0xd8682bfa6918b0174f287b888e765b9a1b4dc9c3": ["8937"]}, + ], +) +```` + +#### `get_nft_metadata` + +Get metadata of NFT. + +````python3 +nfts = ankr_w3.nft.get_nft_metadata( + blockchain="eth", + contract_address="0x4100670ee2f8aef6c47a4ed13c7f246e621228ec", + token_id="4", +) +```` + +#### `get_token_holders` + +Get holders of a token. + +````python3 +holders = ankr_w3.token.get_token_holders( + blockchain="bsc", + contract_address="0xf307910A4c7bbc79691fD374889b36d8531B08e3", + limit=10, +) +```` + +#### `get_token_holders_count_history` + +Get token holders count daily history. + +````python3 +daily_holders_history = ankr_w3.token..get_token_holders_count_history( + blockchain="bsc", + contract_address="0xf307910A4c7bbc79691fD374889b36d8531B08e3", + limit=10, # last 10 days history +) +```` + +#### `get_token_holders_count` + +Get token holders count at the latest block. + +````python3 +holders_count = ankr_w3.token..get_token_holders_count( + blockchain="bsc", + contract_address="0xf307910A4c7bbc79691fD374889b36d8531B08e3", +) +```` + +#### `get_account_balance` + +Get account assets. + +````python3 +assets = ankr_w3.token..get_account_balance( + wallet_address="0x77A859A53D4de24bBC0CC80dD93Fbe391Df45527", + blockchain=["eth", "bsc"], +) +```` #### `get_logs` Get logs matching the filter. ```python3 -logs = ankr_api.get_logs( +logs = ankr_w3.query.get_logs( blockchain="eth", from_block="0xdaf6b1", to_block=14350010, @@ -76,7 +186,7 @@ logs = ankr_api.get_logs( Query data about blocks within a specified range. ```python3 -blocks = ankr_api.get_blocks( +blocks = ankr_w3.query.get_blocks( blockchain="eth", from_block=14500001, to_block=14500001, @@ -87,24 +197,23 @@ blocks = ankr_api.get_blocks( ) ``` -#### `get_nfts` +#### `get_transaction` -Get data about all the NFTs (collectibles) owned by a wallet. +Get Transaction by hash. ````python3 -nfts = ankr_api.get_nfts( - blockchain="eth", - wallet_address="0x0E11A192d574b342C51be9e306694C41547185DD", - filter=[ - {"0x700b4b9f39bb1faf5d0d16a20488f2733550bff4": []}, - {"0xd8682bfa6918b0174f287b888e765b9a1b4dc9c3": ["8937"]}, - ], +tx = ankr_w3.query.get_transaction( + transaction_hash="0x82c13aaac6f0b6471afb94a3a64ae89d45baa3608ad397621dbb0d847f51196f", + include_logs=True, + decode_logs=True, + decode_tx_data=True, ) ```` ### About API keys -For now, Ankr is offering _free_ access to these APIs with no request limits i.e. you don't need an API key at this time. +For now, Ankr is offering _free_ access to these APIs with no request limits i.e. you don't need an API key at this +time. Later on, these APIs will become a part of Ankr Protocol's [Premium Plan](https://www.ankr.com/protocol/plan/). diff --git a/ankr/__init__.py b/ankr/__init__.py index 515cfcd..dc77ad5 100644 --- a/ankr/__init__.py +++ b/ankr/__init__.py @@ -1,3 +1,4 @@ from __future__ import annotations -from ankr.advanced_api import AnkrAdvancedAPI +from ankr.advanced_apis import AnkrAdvancedAPI +from ankr.web3 import AnkrWeb3 diff --git a/ankr/advanced_api.py b/ankr/advanced_api.py deleted file mode 100644 index 30bfc52..0000000 --- a/ankr/advanced_api.py +++ /dev/null @@ -1,88 +0,0 @@ -from __future__ import annotations - -from typing import Any, Dict, Iterable, List, Optional - -from ankr import types -from ankr.provider import AnkrProvider - - -class AnkrAdvancedAPI: - def __init__( - self, - api_key: Optional[str] = None, - endpoint_uri: Optional[str] = None, - ) -> None: - self.provider = AnkrProvider(api_key or "", endpoint_uri) - - def get_logs( - self, - blockchain: types.BlockchainNames, - from_block: Optional[types.BlockNumber] = None, - to_block: Optional[types.BlockNumber] = None, - address: Optional[types.AddressOrAddresses] = None, - topics: Optional[types.Topics] = None, - decode_logs: Optional[bool] = None, - **kwargs: Any, - ) -> Iterable[types.Log]: - for reply in self.provider.call_method_paginated( - "ankr_getLogs", - types.GetLogsRequest( - blockchain=blockchain, - from_block=from_block, - to_block=to_block, - address=address, - topics=topics, - decode_logs=decode_logs, - **kwargs, - ), - types.GetLogsReply, - ): - yield from reply.logs - - def get_blocks( - self, - blockchain: types.BlockchainName, - from_block: Optional[types.BlockNumber] = None, - to_block: Optional[types.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, - **kwargs: Any, - ) -> List[types.Block]: - reply = self.provider.call_method( - "ankr_getBlocks", - types.GetBlocksRequest( - blockchain=blockchain, - from_block=from_block, - to_block=to_block, - desc_order=desc_order, - include_logs=include_logs, - include_txs=include_txs, - decode_logs=decode_logs, - decode_tx_data=decode_tx_data, - **kwargs, - ), - types.GetBlocksReply, - ) - return reply.blocks - - def get_nfts( - self, - blockchain: types.BlockchainNames, - wallet_address: str, - filter: Optional[List[Dict[str, List[str]]]] = None, - **kwargs: Any, - ) -> Iterable[types.Nft]: - for reply in self.provider.call_method_paginated( - "ankr_getNFTsByOwner", - types.GetNFTsByOwnerRequest( - blockchain=blockchain, - wallet_address=wallet_address, - filter=filter, - **kwargs, - ), - types.GetNFTsByOwnerReply, - ): - yield from reply.assets diff --git a/ankr/advanced_apis.py b/ankr/advanced_apis.py new file mode 100644 index 0000000..aeb9596 --- /dev/null +++ b/ankr/advanced_apis.py @@ -0,0 +1,233 @@ +from __future__ import annotations + +from typing import Any, Dict, Iterable, List, Optional + +from ankr import types +from ankr.exceptions import APIError +from ankr.providers import MultichainHTTPProvider + + +class AnkrMultichainAPI: + def __init__( + self, + api_key: Optional[str] = None, + endpoint_uri: Optional[str] = None, + ) -> None: + self.provider = MultichainHTTPProvider(api_key or "", endpoint_uri) + + +class AnkrQueryAPI(AnkrMultichainAPI): + def get_logs( + self, + blockchain: types.BlockchainNames, + from_block: Optional[types.BlockNumber] = None, + to_block: Optional[types.BlockNumber] = None, + address: Optional[types.AddressOrAddresses] = None, + topics: Optional[types.Topics] = None, + decode_logs: Optional[bool] = None, + limit: Optional[int] = None, + **kwargs: Any, + ) -> Iterable[types.Log]: + for log in self.provider.call_method_paginated( + "ankr_getLogs", + types.GetLogsRequest( + blockchain=blockchain, + from_block=from_block, + to_block=to_block, + address=address, + topics=topics, + decode_logs=decode_logs, + **kwargs, + ), + reply_type=types.GetLogsReply, + iterable_name="logs", + iterable_type=types.Log, + limit=limit, + ): + yield log + + def get_blocks( + self, + blockchain: types.BlockchainName, + from_block: Optional[types.BlockNumber] = None, + to_block: Optional[types.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, + **kwargs: Any, + ) -> List[types.Block]: + reply = self.provider.call_method( + "ankr_getBlocks", + types.GetBlocksRequest( + blockchain=blockchain, + from_block=from_block, + to_block=to_block, + desc_order=desc_order, + include_logs=include_logs, + include_txs=include_txs, + decode_logs=decode_logs, + decode_tx_data=decode_tx_data, + **kwargs, + ), + types.GetBlocksReply, + ) + return reply.blocks + + 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[types.Transaction]: + reply = self.provider.call_method( + "ankr_getTransactionsByHash", + types.GetTransactionsByHashRequest( + blockchain=blockchain, + transaction_hash=transaction_hash, + include_logs=include_logs, + decode_logs=decode_logs, + decode_tx_data=decode_tx_data, + **kwargs, + ), + types.GetTransactionsByHashReply, + ) + if reply.transactions: + return reply.transactions[0] + + +class AnkrTokenAPI(AnkrMultichainAPI): + def get_token_holders( + self, + blockchain: types.BlockchainName, + contract_address: str, + limit: Optional[int] = None, + **kwargs: Any, + ) -> Iterable[types.HolderBalance]: + for holder in self.provider.call_method_paginated( + "ankr_getTokenHolders", + types.GetTokenHoldersRequest( + blockchain=blockchain, + contract_address=contract_address, + reply_type=types.GetLogsReply, + **kwargs, + ), + types.GetTokenHoldersReply, + iterable_name="holders", + iterable_type=types.HolderBalance, + limit=limit, + ): + yield holder + + def get_token_holders_count_history( + self, + blockchain: types.BlockchainName, + contract_address: str, + limit: Optional[int] = None, + **kwargs: Any, + ) -> Iterable[types.DailyHolderCount]: + for daily_holder_count in self.provider.call_method_paginated( + "ankr_getTokenHoldersCount", + types.GetTokenHoldersCountRequest( + blockchain=blockchain, + contract_address=contract_address, + **kwargs, + ), + types.GetTokenHoldersCountReply, + iterable_name="holder_count_history", + iterable_type=types.DailyHolderCount, + limit=limit, + ): + yield daily_holder_count + + def get_token_holders_count( + self, + blockchain: types.BlockchainName, + contract_address: str, + **kwargs: Any, + ) -> types.DailyHolderCount: + reply = self.provider.call_method( + rpc="ankr_getTokenHoldersCount", + request=types.GetTokenHoldersCountRequest( + blockchain=blockchain, + contract_address=contract_address, + page_size=1, + **kwargs, + ), + reply_type=types.GetTokenHoldersCountReply, + ) + if len(reply.holder_count_history) < 1: + raise APIError("no token holders count found") + return reply.holder_count_history[0] + + def get_account_balance( + self, + wallet_address: str, + blockchain: Optional[types.BlockchainNames] = None, + limit: Optional[int] = None, + **kwargs: Any, + ) -> Iterable[types.Balance]: + for asset in self.provider.call_method_paginated( + rpc="ankr_getAccountBalance", + request=types.GetAccountBalanceRequest( + blockchain=blockchain, + wallet_address=wallet_address, + **kwargs, + ), + reply_type=types.GetAccountBalanceReply, + iterable_name="assets", + iterable_type=types.Balance, + limit=limit, + ): + yield asset + + +class AnkrNFTAPI(AnkrMultichainAPI): + def get_nfts( + self, + blockchain: types.BlockchainNames, + wallet_address: str, + filter: Optional[List[Dict[str, List[str]]]] = None, + limit: Optional[int] = None, + **kwargs: Any, + ) -> Iterable[types.Nft]: + for nft in self.provider.call_method_paginated( + "ankr_getNFTsByOwner", + types.GetNFTsByOwnerRequest( + blockchain=blockchain, + wallet_address=wallet_address, + filter=filter, + **kwargs, + ), + types.GetNFTsByOwnerReply, + iterable_name="assets", + iterable_type=types.Nft, + limit=limit, + ): + yield nft + + def get_nft_metadata( + self, + blockchain: types.BlockchainName, + contract_address: str, + token_id: str, + **kwargs: Any, + ) -> types.GetNFTMetadataReply: + return self.provider.call_method( + "ankr_getNFTMetadata", + types.GetNFTMetadataRequest( + blockchain=blockchain, + contract_address=contract_address, + token_id=token_id, + **kwargs, + ), + types.GetNFTMetadataReply, + ) + + +class AnkrAdvancedAPI(AnkrQueryAPI, AnkrTokenAPI, AnkrNFTAPI): + ... diff --git a/ankr/exceptions.py b/ankr/exceptions.py index 05349ad..666b210 100644 --- a/ankr/exceptions.py +++ b/ankr/exceptions.py @@ -3,6 +3,6 @@ from web3.types import RPCError -class AdvancedAPIException(Exception): +class APIError(Exception): def __init__(self, error: RPCError | str) -> None: super().__init__(f"failed to handle request, {error}") diff --git a/ankr/provider.py b/ankr/provider.py deleted file mode 100644 index ecd8c2a..0000000 --- a/ankr/provider.py +++ /dev/null @@ -1,63 +0,0 @@ -from __future__ import annotations - -from typing import Any, Iterable, Type, TypeVar - -from eth_typing import URI -from web3 import HTTPProvider -from web3.types import RPCEndpoint, RPCResponse - -from ankr import types -from ankr.exceptions import AdvancedAPIException - -TRequest = TypeVar("TRequest", bound=types.RPCModel) -TReply = TypeVar("TReply", bound=types.RPCModel) -TRequestPaginated = TypeVar("TRequestPaginated", bound=types.RPCRequestPaginated) -TReplyPaginated = TypeVar("TReplyPaginated", bound=types.RPCReplyPaginated) - - -class AnkrProvider(HTTPProvider): - def __init__( - self, - api_key: str = "", - endpoint_uri: URI | str | None = None, - request_kwargs: Any | None = None, - session: Any = None, - ) -> None: - if endpoint_uri is None: - endpoint_uri = "https://rpc.ankr.com/multichain/" - super().__init__(endpoint_uri + api_key, request_kwargs) - - def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse: - response = super().make_request(method, params) - if response.get("error"): - raise AdvancedAPIException(response["error"]) - if "result" not in response: - raise AdvancedAPIException("returned no result") - return response - - def call_method( - self, - rpc: str, - request: TRequest, - reply_type: Type[TReply], - ) -> TReply: - request_dict = request.dict(by_alias=True, exclude_none=True) - response = self.make_request(RPCEndpoint(rpc), request_dict) - reply = reply_type(**response["result"]) - return reply - - def call_method_paginated( - self, - rpc: str, - request: TRequestPaginated, - reply_type: Type[TReplyPaginated], - ) -> Iterable[TReplyPaginated]: - request_dict = request.dict(by_alias=True, exclude_none=True) - response = self.make_request(RPCEndpoint(rpc), request_dict) - reply = reply_type(**response["result"]) - - yield reply - - if reply.next_page_token: - request.page_token = reply.next_page_token - yield from self.call_method_paginated(RPCEndpoint(rpc), request, reply_type) diff --git a/ankr/providers.py b/ankr/providers.py new file mode 100644 index 0000000..5e7d0e3 --- /dev/null +++ b/ankr/providers.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +from typing import Any, Iterable, List, Optional, Protocol, Type, TypeVar, Union + +from eth_typing import URI +from web3 import HTTPProvider +from web3.types import RPCEndpoint, RPCResponse + +from ankr import types +from ankr.exceptions import APIError + +TRequest = TypeVar("TRequest", bound=types.RPCModel) +TReply = TypeVar("TReply", bound=types.RPCModel) +TRequestPaginated = TypeVar("TRequestPaginated", bound=types.RPCRequestPaginated) +TReplyPaginated = TypeVar("TReplyPaginated", bound=types.RPCReplyPaginated) + + +class MultichainHTTPProvider(HTTPProvider): + def __init__( + self, + api_key: str = "", + endpoint_uri: Optional[Union[URI, str]] = None, + request_kwargs: Optional[Any] = None, + session: Optional[Any] = None, + ) -> None: + if endpoint_uri is None: + endpoint_uri = "https://rpc.ankr.com/multichain/" + super().__init__(endpoint_uri + api_key, request_kwargs, session) + + def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse: + response = super().make_request(method, params) + if response.get("error"): + raise APIError(response["error"]) + if "result" not in response: + raise APIError("returned no result") + return response + + def call_method( + self, + rpc: str, + request: TRequest, + reply_type: Type[TReply], + ) -> TReply: + request_dict = request.dict(by_alias=True, exclude_none=True) + response = self.make_request(RPCEndpoint(rpc), request_dict) + reply = reply_type(**response["result"]) + return reply + + def call_method_paginated( + self, + rpc: str, + request: TRequestPaginated, + reply_type: Type[TReplyPaginated], + iterable_name: str, + iterable_type: Type[TReply], + limit: Optional[int] = None, + ) -> Iterable[TReply]: + request_dict = request.dict(by_alias=True, exclude_none=True) + response = self.make_request(RPCEndpoint(rpc), request_dict) + reply = reply_type(**response["result"]) + + items: List[TReply] = getattr(reply, iterable_name, []) + + if limit: + if limit <= len(items): + yield from items[:limit] + return + limit -= len(items) + + yield from items + + if reply.next_page_token: + request.page_token = reply.next_page_token + yield from self.call_method_paginated( + RPCEndpoint(rpc), + request, + reply_type, + iterable_name, + iterable_type, + limit, + ) + + +class TProviderConstructor(Protocol): + def __call__( + self, api_key: Optional[str] = None, request_kwargs: Optional[Any] = None + ) -> HTTPProvider: + ... + + +def http_provider_constructor(url: str) -> TProviderConstructor: + def init( + api_key: Optional[str] = None, + request_kwargs: Optional[Any] = None, + ) -> HTTPProvider: + if api_key is None: + api_key = "" + return HTTPProvider(url + api_key, request_kwargs) + + return init + + +# EVM +ArbitrumHTTPProvider = http_provider_constructor("https://rpc.ankr.com/arbitrum") +AvalancheHTTPProvider = http_provider_constructor("https://rpc.ankr.com/avalanche") +BscHTTPProvider = http_provider_constructor("https://rpc.ankr.com/bsc") +CeloHTTPProvider = http_provider_constructor("https://rpc.ankr.com/celo") +EthHTTPProvider = http_provider_constructor("https://rpc.ankr.com/eth") +FantomHTTPProvider = http_provider_constructor("https://rpc.ankr.com/fantom") +GnosisHTTPProvider = http_provider_constructor("https://rpc.ankr.com/gnosis") +HarmonyHTTPProvider = http_provider_constructor("https://rpc.ankr.com/harmony") +IotexHTTPProvider = http_provider_constructor("https://rpc.ankr.com/iotex") +MoonbeamHTTPProvider = http_provider_constructor("https://rpc.ankr.com/moonbeam") +NervosHTTPProvider = http_provider_constructor("https://rpc.ankr.com/nervos") +OptimismHTTPProvider = http_provider_constructor("https://rpc.ankr.com/optimism") +PolygonHTTPProvider = http_provider_constructor("https://rpc.ankr.com/polygon") +SyscoinHTTPProvider = http_provider_constructor("https://rpc.ankr.com/syscoin") + +# Non-EVM +NearHTTPProvider = http_provider_constructor("https://rpc.ankr.com/near") +SolanaHTTPProvider = http_provider_constructor("https://rpc.ankr.com/solana") diff --git a/ankr/types.py b/ankr/types.py index 9f4f212..252e67f 100644 --- a/ankr/types.py +++ b/ankr/types.py @@ -8,7 +8,7 @@ from pydantic import BaseModel -class BlockchainName(str, enum.Enum): +class Blockchain(str, enum.Enum): ETH = "eth" BSC = "bsc" POLYGON = "polygon" @@ -18,12 +18,19 @@ class BlockchainName(str, enum.Enum): SYSCOIN = "syscoin" +class NftContractType(str, enum.Enum): + ERC721 = "ERC721" + ERC1155 = "ERC1155" + UNDEFINED = "UNDEFINED" + + class BlockNumberName(str, enum.Enum): latest = "latest" earliest = "earliest" -BlockchainNames = Union[BlockchainName, List[BlockchainName], str] +BlockchainName = Union[Blockchain, str] +BlockchainNames = Union[BlockchainName, List[BlockchainName]] BlockNumber = Union[int, str, BlockNumberName] AddressOrAddresses = Union[str, List[str]] Topics = Union[str, List[Union[str, List[str]]]] @@ -94,7 +101,7 @@ class NftAttributes(RPCModel): image_url: str name: str description: str - contract_type: int + contract_type: NftContractType traits: Optional[List[Attribute]] = None @@ -102,7 +109,7 @@ class NftMetadata(RPCModel): blockchain: BlockchainName contract_address: str token_id: str - contract_type: int + contract_type: NftContractType class GetNFTMetadataReply(RPCModel): @@ -132,7 +139,7 @@ class GetAccountBalanceReply(RPCReplyPaginated): class GetAccountBalanceRequest(RPCRequestPaginated): - blockchain: Optional[Union[BlockchainName, List[BlockchainName]]] + blockchain: Optional[BlockchainNames] wallet_address: str page_token: Optional[str] = None page_size: Optional[int] = None @@ -247,7 +254,7 @@ class GetLogsReply(RPCReplyPaginated): class GetLogsRequest(RPCRequestPaginated): - blockchain: Union[BlockchainName, List[BlockchainName]] + blockchain: BlockchainNames from_block: Optional[BlockNumber] = None to_block: Optional[BlockNumber] = None address: Optional[Union[str, List[str]]] = None @@ -347,7 +354,7 @@ class GetBlocksReply(RPCModel): class GetTransactionsByHashRequest(RPCModel): - blockchain: Optional[Union[BlockchainName, List[BlockchainName]]] + blockchain: Optional[BlockchainNames] transaction_hash: str include_logs: Optional[bool] = None decode_logs: Optional[bool] = None diff --git a/ankr/web3.py b/ankr/web3.py new file mode 100644 index 0000000..1aa9094 --- /dev/null +++ b/ankr/web3.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from typing import Any, Dict, Optional, Sequence, Type, Union, cast + +from ens import ENS +from web3 import Web3 +from web3._utils.empty import empty +from web3.eth import Eth +from web3.middleware import geth_poa_middleware +from web3.module import Module + +from ankr.advanced_apis import AnkrNFTAPI, AnkrQueryAPI, AnkrTokenAPI +from ankr.providers import ( + ArbitrumHTTPProvider, + AvalancheHTTPProvider, + BscHTTPProvider, + CeloHTTPProvider, + EthHTTPProvider, + FantomHTTPProvider, + GnosisHTTPProvider, + HarmonyHTTPProvider, + IotexHTTPProvider, + MoonbeamHTTPProvider, + NervosHTTPProvider, + OptimismHTTPProvider, + PolygonHTTPProvider, + SyscoinHTTPProvider, + TProviderConstructor, +) + + +class _Web3NamedMeta(type): + def __new__(metacls, name, bases, namespace, **kw): # type: ignore + return super().__new__(metacls, "Web3", bases, namespace, **kw) + + +class AnkrWeb3(Web3, metaclass=_Web3NamedMeta): + query: AnkrQueryAPI + token: AnkrTokenAPI + nft: AnkrNFTAPI + + eth: Eth + arbitrum: Eth + avalanche: Eth + bsc: Eth + celo: Eth + fantom: Eth + gnosis: Eth + harmony: Eth + iotex: Eth + moonbeam: Eth + nervos: Eth + optimism: Eth + polygon: Eth + syscoin: Eth + + def __init__( + self, + api_key: Optional[str] = None, + request_kwargs: Optional[Any] = None, + middlewares: Optional[Sequence[Any]] = None, + modules: Optional[Dict[str, Union[Type[Module], Sequence[Any]]]] = None, + external_modules: Optional[ + Dict[str, Union[Type[Module], Sequence[Any]]] + ] = None, + ens: ENS = cast(ENS, empty), + ) -> None: + self.__api_key = api_key + self.__request_kwargs = request_kwargs + self.__middlewares = middlewares + self.__modules = modules + self.__external_modules = external_modules + self.__ens = ens + + self.query = AnkrQueryAPI(api_key) + self.token = AnkrTokenAPI(api_key) + self.nft = AnkrNFTAPI(api_key) + + eth_provider = EthHTTPProvider(api_key, request_kwargs) + super().__init__(eth_provider, middlewares, modules, external_modules, ens) + self.arbitrum = self.__new_evm_chain(ArbitrumHTTPProvider) + self.avalanche = self.__new_evm_chain(AvalancheHTTPProvider) + self.bsc = self.__new_evm_chain(BscHTTPProvider) + self.celo = self.__new_evm_chain(CeloHTTPProvider) + self.fantom = self.__new_evm_chain(FantomHTTPProvider) + self.gnosis = self.__new_evm_chain(GnosisHTTPProvider) + self.harmony = self.__new_evm_chain(HarmonyHTTPProvider) + self.iotex = self.__new_evm_chain(IotexHTTPProvider) + self.moonbeam = self.__new_evm_chain(MoonbeamHTTPProvider) + self.nervos = self.__new_evm_chain(NervosHTTPProvider) + self.optimism = self.__new_evm_chain(OptimismHTTPProvider) + self.polygon = self.__new_evm_chain(PolygonHTTPProvider) + self.syscoin = self.__new_evm_chain(SyscoinHTTPProvider) + + def __new_evm_chain(self, provider: TProviderConstructor) -> Eth: + w3 = Web3( + provider(self.__api_key, self.__request_kwargs), + self.__middlewares, + self.__modules, + self.__external_modules, + self.__ens, + ) + w3.middleware_onion.inject(geth_poa_middleware, layer=0) + return w3.eth diff --git a/setup.cfg b/setup.cfg index 4f0c7ba..3544177 100644 --- a/setup.cfg +++ b/setup.cfg @@ -8,9 +8,6 @@ allow_redefinition = True follow_imports = skip exclude = env|venv|venv.*|tests|test_* -[mypy-*.tests.*] -ignore_errors = True - [flake8] max-complexity = 8 max-annotations-complexity = 4 @@ -18,4 +15,8 @@ max-line-length = 120 max-function-length = 100 exclude = env,venv,pytest.ini per-file-ignores = - __init__.py: F401 \ No newline at end of file + __init__.py: F401 + +[tool:pytest] +markers = + webtest: mark a test as a webtest. \ No newline at end of file diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..6fb6e40 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +import datetime + +import pytest + +from ankr.advanced_apis import AnkrAdvancedAPI +from ankr.types import Blockchain, NftContractType + + +def test_client_api_key() -> None: + client = AnkrAdvancedAPI() + client_with_key = AnkrAdvancedAPI("my-test-api-key") + + assert client.provider.endpoint_uri == "https://rpc.ankr.com/multichain/" + assert ( + client_with_key.provider.endpoint_uri + == "https://rpc.ankr.com/multichain/my-test-api-key" + ) + + +@pytest.mark.webtest +def test_get_logs() -> None: + client = AnkrAdvancedAPI() + logs = list( + client.get_logs( + blockchain=Blockchain.ETH, + from_block="0xdaf6b1", + to_block=14350010, + address=["0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"], + topics=[ + [], + ["0x000000000000000000000000def1c0ded9bec7f1a1670819833240f027b25eff"], + ], + decode_logs=True, + ) + ) + + assert len(logs) == 18 + assert logs[0].address == "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2" + assert logs[0].event + assert logs[0].event.name == "Deposit" + + +@pytest.mark.webtest +def test_get_blocks() -> None: + client = AnkrAdvancedAPI() + blocks = client.get_blocks( + blockchain=Blockchain.ETH, + from_block=14500001, + to_block=14500001, + desc_order=True, + include_logs=True, + include_txs=True, + decode_logs=True, + ) + + assert len(blocks) == 1 + assert blocks[0].transactions + assert len(blocks[0].transactions) == 99 + assert len(blocks[0].transactions[6].logs) == 1 + + +@pytest.mark.webtest +def test_get_nfts() -> None: + client = AnkrAdvancedAPI() + nfts = list( + client.get_nfts( + blockchain=Blockchain.ETH, + wallet_address="0x0E11A192d574b342C51be9e306694C41547185DD", + filter=[ + {"0x700b4b9f39bb1faf5d0d16a20488f2733550bff4": []}, + {"0xd8682bfa6918b0174f287b888e765b9a1b4dc9c3": ["8937"]}, + ], + ) + ) + + assert len(nfts) > 0 + assert nfts[0].blockchain == Blockchain.ETH + assert nfts[0].traits + assert len(nfts[0].traits) > 0 + + +@pytest.mark.webtest +def test_get_nft_metadata() -> None: + client = AnkrAdvancedAPI() + reply = client.get_nft_metadata( + blockchain="eth", + contract_address="0x4100670ee2f8aef6c47a4ed13c7f246e621228ec", + token_id="4", + ) + + assert reply.metadata + assert reply.metadata.blockchain == "eth" + assert reply.metadata.contract_type == NftContractType.ERC1155 + assert reply.attributes + assert reply.attributes.name == "Overleveraged" + + +@pytest.mark.webtest +def test_get_transactions() -> None: + client = AnkrAdvancedAPI() + tx = client.get_transaction( + transaction_hash="0x82c13aaac6f0b6471afb94a3a64ae89d45baa3608ad397621dbb0d847f51196f", + include_logs=True, + decode_logs=True, + decode_tx_data=True, + ) + + assert tx + assert ( + tx.transaction_hash + == "0x82c13aaac6f0b6471afb94a3a64ae89d45baa3608ad397621dbb0d847f51196f" + ) + assert tx.to_address == "0x98767abab06e45a181ab73ae4cd0fecd0fbd0cd0" + assert tx.from_address == "0x64aa6f93e0e1f49ff4958990c40d4bf17dafc0eb" + assert tx.logs + assert tx.logs[0].event + assert tx.logs[0].event.name == "Transfer" + + +@pytest.mark.webtest +def test_get_token_holders() -> None: + client = AnkrAdvancedAPI() + holders = list( + client.get_token_holders( + blockchain="bsc", + contract_address="0xf307910A4c7bbc79691fD374889b36d8531B08e3", + limit=10, + ) + ) + + assert len(holders) == 10 + assert holders[0].holder_address.startswith("0x") + assert "." in holders[0].balance + assert holders[0].balance_raw_integer.isnumeric() + + +@pytest.mark.webtest +def test_get_token_holders_count_history() -> None: + client = AnkrAdvancedAPI() + daily_holders_counts = list( + client.get_token_holders_count_history( + blockchain="bsc", + contract_address="0xf307910A4c7bbc79691fD374889b36d8531B08e3", + limit=10, + ) + ) + + assert len(daily_holders_counts) == 10 + assert daily_holders_counts[0].holder_count > 0 + datetime.datetime.strptime( + daily_holders_counts[0].last_updated_at, "%Y-%m-%dT%H:%M:%SZ" + ) + + +@pytest.mark.webtest +def test_get_token_holders_count() -> None: + client = AnkrAdvancedAPI() + holders_count = client.get_token_holders_count( + blockchain="bsc", + contract_address="0xf307910A4c7bbc79691fD374889b36d8531B08e3", + ) + + assert holders_count + assert holders_count.holder_count > 0 + datetime.datetime.strptime(holders_count.last_updated_at, "%Y-%m-%dT%H:%M:%SZ") + + +@pytest.mark.webtest +def test_get_account_balance() -> None: + client = AnkrAdvancedAPI() + assets = list( + client.get_account_balance( + wallet_address="0x77A859A53D4de24bBC0CC80dD93Fbe391Df45527", + blockchain=["eth", "bsc"], + ) + ) + + assert assets + assert len(assets) > 0 diff --git a/tests/test_provider.py b/tests/test_provider.py deleted file mode 100644 index 2b8d459..0000000 --- a/tests/test_provider.py +++ /dev/null @@ -1,71 +0,0 @@ -from __future__ import annotations - -from ankr import AnkrAdvancedAPI, types - - -def test_provider_api_key(): - advanced_api = AnkrAdvancedAPI() - assert advanced_api.provider.endpoint_uri == "https://rpc.ankr.com/multichain/" - - advanced_api = AnkrAdvancedAPI("my-test-api-key") - assert ( - advanced_api.provider.endpoint_uri - == "https://rpc.ankr.com/multichain/my-test-api-key" - ) - - -def test_get_logs(): - advanced_api = AnkrAdvancedAPI() - logs = list( - advanced_api.get_logs( - blockchain=types.BlockchainName.ETH, - from_block="0xdaf6b1", - to_block=14350010, - address=["0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"], - topics=[ - [], - ["0x000000000000000000000000def1c0ded9bec7f1a1670819833240f027b25eff"], - ], - decode_logs=True, - ) - ) - - assert len(logs) == 18 - assert logs[0].address == "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2" - assert logs[0].event.name == "Deposit" - - -def test_get_blocks(): - advanced_api = AnkrAdvancedAPI() - blocks = advanced_api.get_blocks( - blockchain=types.BlockchainName.ETH, - from_block=14500001, - to_block=14500001, - desc_order=True, - include_logs=True, - include_txs=True, - decode_logs=True, - ) - - assert len(blocks) == 1 - assert blocks[0].transactions - assert len(blocks[0].transactions) == 99 - assert len(blocks[0].transactions[6].logs) == 1 - - -def test_get_nfts(): - advanced_api = AnkrAdvancedAPI() - nfts = list( - advanced_api.get_nfts( - blockchain=types.BlockchainName.ETH, - wallet_address="0x0E11A192d574b342C51be9e306694C41547185DD", - filter=[ - {"0x700b4b9f39bb1faf5d0d16a20488f2733550bff4": []}, - {"0xd8682bfa6918b0174f287b888e765b9a1b4dc9c3": ["8937"]}, - ], - ) - ) - - assert len(nfts) > 0 - assert nfts[0].blockchain == types.BlockchainName.ETH - assert len(nfts[0].traits) > 0 diff --git a/tests/test_providers.py b/tests/test_providers.py new file mode 100644 index 0000000..9e306e2 --- /dev/null +++ b/tests/test_providers.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +from ankr.providers import ( + ArbitrumHTTPProvider, + AvalancheHTTPProvider, + BscHTTPProvider, + CeloHTTPProvider, + EthHTTPProvider, + FantomHTTPProvider, + GnosisHTTPProvider, + HarmonyHTTPProvider, + IotexHTTPProvider, + MoonbeamHTTPProvider, + MultichainHTTPProvider, + NearHTTPProvider, + NervosHTTPProvider, + OptimismHTTPProvider, + PolygonHTTPProvider, + SolanaHTTPProvider, + SyscoinHTTPProvider, +) + + +def test_provider_api_key() -> None: + provider = MultichainHTTPProvider() + provider_with_key = MultichainHTTPProvider("my-test-api-key") + + assert provider.endpoint_uri == "https://rpc.ankr.com/multichain/" + assert ( + provider_with_key.endpoint_uri + == "https://rpc.ankr.com/multichain/my-test-api-key" + ) + + +def test_chain_providers() -> None: + assert ArbitrumHTTPProvider().endpoint_uri == "https://rpc.ankr.com/arbitrum" + assert AvalancheHTTPProvider().endpoint_uri == "https://rpc.ankr.com/avalanche" + assert BscHTTPProvider().endpoint_uri == "https://rpc.ankr.com/bsc" + assert CeloHTTPProvider().endpoint_uri == "https://rpc.ankr.com/celo" + assert EthHTTPProvider().endpoint_uri == "https://rpc.ankr.com/eth" + assert FantomHTTPProvider().endpoint_uri == "https://rpc.ankr.com/fantom" + assert GnosisHTTPProvider().endpoint_uri == "https://rpc.ankr.com/gnosis" + assert HarmonyHTTPProvider().endpoint_uri == "https://rpc.ankr.com/harmony" + assert IotexHTTPProvider().endpoint_uri == "https://rpc.ankr.com/iotex" + assert MoonbeamHTTPProvider().endpoint_uri == "https://rpc.ankr.com/moonbeam" + assert NearHTTPProvider().endpoint_uri == "https://rpc.ankr.com/near" + assert NervosHTTPProvider().endpoint_uri == "https://rpc.ankr.com/nervos" + assert OptimismHTTPProvider().endpoint_uri == "https://rpc.ankr.com/optimism" + assert PolygonHTTPProvider().endpoint_uri == "https://rpc.ankr.com/polygon" + assert SolanaHTTPProvider().endpoint_uri == "https://rpc.ankr.com/solana" + assert SyscoinHTTPProvider().endpoint_uri == "https://rpc.ankr.com/syscoin" diff --git a/tests/test_web3.py b/tests/test_web3.py new file mode 100644 index 0000000..4f8b1c7 --- /dev/null +++ b/tests/test_web3.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import pytest + +from ankr.web3 import AnkrWeb3 + + +@pytest.mark.webtest +@pytest.mark.parametrize( + "blockchain", + [ + "eth", + "arbitrum", + "avalanche", + "bsc", + "celo", + "fantom", + "gnosis", + "harmony", + "iotex", + "moonbeam", + "nervos", + "optimism", + "polygon", + "syscoin", + ], +) +def test_ankr_web3(blockchain: str) -> None: + w3 = AnkrWeb3() + + block = getattr(w3, blockchain).get_block("latest") + + assert block + assert block.get("number")