pre-release draft

Author: fasashen

.github/workflows/publish.yaml

@@ -0,0 +1,30 @@
+name: publish
+
+on:
+  release:
+    types: [created]
+
+jobs:
+  publish:
+    runs-on: "ubuntu-latest"
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v2
+
+      - name: Set up python
+        id: setup-python
+        uses: actions/setup-python@v2
+        with:
+          python-version: "3.10"
+
+      - name: Install Poetry
+        uses: snok/install-poetry@v1
+
+      - name: Build library
+        run: poetry build
+
+      - name: Authenticate in PyPi
+        run: poetry config http-basic.pypi __token__ ${{ secrets.PYPI_TOKEN }}
+
+      - name: Publish library
+        run: poetry publish

.github/workflows/test.yaml

@@ -77,5 +77,4 @@ jobs:
       - name: Run tests
         run: |
           source .venv/bin/activate
-          pytest tests/
-          coverage report
+          pytest

.gitignore

@@ -127,3 +127,6 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# Pycharm
+.idea/

README.md

@@ -0,0 +1,110 @@
+# ⚓️ Ankr Python SDK
+
+Compact Python library for interacting with Ankr's [Advanced APIs](https://www.ankr.com/advanced-api/).
+
+## Get started in 2 minutes
+
+#### 1. Install the package from PyPi
+
+```bash
+pip install ankr-sdk
+```
+
+#### 2. Initialize the SDK
+
+```python3
+from ankr import AnkrAdvancedAPI, types
+
+ankr_api = AnkrAdvancedAPI()
+```
+
+####3. Use the sdk and call one of the supported methods
+
+```python3
+from ankr.types import BlockchainName
+
+nfts = ankr_api.get_nfts(
+    blockchain=BlockchainName.ETH,
+    wallet_address="0x0E11A192d574b342C51be9e306694C41547185DD",
+    filter=[
+        {"0x700b4b9f39bb1faf5d0d16a20488f2733550bff4": []},
+        {"0xd8682bfa6918b0174f287b888e765b9a1b4dc9c3": ["8937"]},
+    ],
+)
+```
+
+## Supported chains
+
+`ankr-sdk` supports the following chains at this time:
+
+- ETH: `"eth"`
+- BSC: `"bsc"`
+- Polygon: `"polygon"`
+- Fantom: `"fantom"`
+- Arbitrum: `"arbitrum"`
+- Avalanche: `"avalanche"`
+- Syscoin NEVM: `"syscoin"`
+
+## Available methods
+
+`ankr-sdk` supports the following methods:
+
+- [`get_nfts`](#get_nfts)
+- [`get_logs`](#get_logs)
+- [`get_blocks`](#get_blocks)
+
+#### `get_logs`
+
+Get logs matching the filter.
+
+```python3
+logs = ankr_api.get_logs(
+    blockchain="eth",
+    from_block="0xdaf6b1",
+    to_block=14350010,
+    address=["0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"],
+    topics=[
+        [],
+        ["0x000000000000000000000000def1c0ded9bec7f1a1670819833240f027b25eff"],
+    ],
+    decode_logs=True,
+)
+```
+
+#### `get_blocks`
+
+Query data about blocks within a specified range.
+
+```python3
+blocks = ankr_api.get_blocks(
+    blockchain="eth",
+    from_block=14500001,
+    to_block=14500001,
+    desc_order=True,
+    include_logs=True,
+    include_txs=True,
+    decode_logs=True,
+)
+```
+
+#### `get_nfts`
+
+Get data about all the NFTs (collectibles) owned by a wallet.
+
+````python3
+nfts = ankr_api.get_nfts(
+    blockchain="eth",
+    wallet_address="0x0E11A192d574b342C51be9e306694C41547185DD",
+    filter=[
+        {"0x700b4b9f39bb1faf5d0d16a20488f2733550bff4": []},
+        {"0xd8682bfa6918b0174f287b888e765b9a1b4dc9c3": ["8937"]},
+    ],
+)
+````
+
+
+### 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.
+
+Later on, these APIs will become a part of Ankr Protocol's [Premium Plan](https://www.ankr.com/protocol/plan/).

ankr/__init__.py

@@ -1 +1,3 @@
-from .provider import AnkrAdvancedAPI
+from __future__ import annotations
+
+from ankr.advanced_api import AnkrAdvancedAPI

ankr/advanced_api.py

@@ -0,0 +1,88 @@
+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

ankr/exceptions.py

@@ -1,8 +1,8 @@
-from typing import Union
+from __future__ import annotations
 
 from web3.types import RPCError
 
 
 class AdvancedAPIException(Exception):
-    def __init__(self, error: Union[RPCError, str]):
-        super().__init__(f"Failed to handle request: {error}")
+    def __init__(self, error: RPCError | str) -> None:
+        super().__init__(f"failed to handle request, {error}")

ankr/provider.py

@@ -1,38 +1,27 @@
-from typing import (
-    Optional,
-    Union,
-    Any,
-    List,
-    Iterable,
-    TypeVar,
-    Type,
-    Dict,
-)
+from __future__ import annotations
 
-from eth_typing import (
-    URI,
-)
+from typing import Any, Iterable, Type, TypeVar
+
+from eth_typing import URI
 from web3 import HTTPProvider
-from web3.types import (
-    RPCEndpoint,
-    RPCResponse,
-)
+from web3.types import RPCEndpoint, RPCResponse
 
 from ankr import types
 from ankr.exceptions import AdvancedAPIException
 
-
-TRequestPaginated = TypeVar("TRequestPaginated", bound=types.RequestPaginated)
-TReplyPaginated = TypeVar("TReplyPaginated", bound=types.ReplyPaginated)
+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: Optional[Union[URI, str]] = None,
-        request_kwargs: Optional[Any] = None,
-        session: Optional[Any] = None,
+        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/"
@@ -46,7 +35,18 @@ def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
             raise AdvancedAPIException("returned no result")
         return response
 
-    def make_request_paginated(
+    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,
@@ -60,59 +60,4 @@ def make_request_paginated(
 
         if reply.next_page_token:
             request.page_token = reply.next_page_token
-            yield from self.make_request_paginated(
-                RPCEndpoint(rpc), request, reply_type
-            )
-
-
-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.make_request_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_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.make_request_paginated(
-            "ankr_getNFTsByOwner",
-            types.GetNFTsByOwnerRequest(
-                blockchain=blockchain,
-                wallet_address=wallet_address,
-                filter=filter,
-                **kwargs,
-            ),
-            types.GetNFTsByOwnerReply,
-        ):
-            yield from reply.assets
+            yield from self.call_method_paginated(RPCEndpoint(rpc), request, reply_type)