Merge b4fc650 into 55c80c5

tests/test_api_handler.py

@@ -1,6 +1,8 @@
 import pytest
+import requests_mock
 
 from web3data.chains import Chains
+from web3data.exceptions import APIError
 from web3data.handlers import APIHandler
 from web3data.handlers.address import AddressHandler
 from web3data.handlers.block import BlockHandler
@@ -12,6 +14,16 @@
 
 TEST_KEY = "test-key"
 TEST_ID = "test-id"
+TEST_RPC = {"test": "response"}
+ALLOWED_CHAINS = (Chains.BTC, Chains.ETH, Chains.ETH_RINKEBY)
+NON_RPC_CHAINS = (
+    Chains.AION,
+    Chains.BCH,
+    Chains.BSV,
+    Chains.LTC,
+    Chains.XLM,
+    Chains.ZEC,
+)
 
 
 @pytest.mark.parametrize(
@@ -36,3 +48,32 @@ def test_api_handler_initialized(handler):
     assert isinstance(handler.signature, SignatureHandler)
     assert isinstance(handler.token, TokenHandler)
     assert isinstance(handler.transaction, TransactionHandler)
+
+
+@pytest.mark.parametrize("chain", ALLOWED_CHAINS)
+def test_allowed_rpc(chain):
+    handler = APIHandler(TEST_KEY, TEST_ID, chain)
+    with requests_mock.Mocker() as m:
+        m.register_uri(requests_mock.ANY, requests_mock.ANY, json=TEST_RPC)
+
+        response = handler.rpc("test-method", ["test-param"])
+        assert m.call_count == 1
+        assert response == TEST_RPC
+        assert (
+            m.request_history[0].url
+            == "https://rpc.web3api.io/?x-api-key=test-key"
+        )
+        assert m.request_history[0].method == "POST"
+        assert m.request_history[0].json() == {
+            "id": 1,
+            "jsonrpc": "2.0",
+            "method": "test-method",
+            "params": ["test-param"],
+        }
+
+
+@pytest.mark.parametrize("chain", NON_RPC_CHAINS)
+def test_non_rpc(chain):
+    handler = APIHandler(TEST_KEY, TEST_ID, chain)
+    with pytest.raises(APIError):
+        handler.rpc("test-method", ["test-param"])

tests/test_base_handler.py

@@ -89,10 +89,3 @@ def test_base_handler_valid_response(chain):
 
         assert resp == RESPONSE
         assert_request_mock(m)
-
-
-@pytest.mark.parametrize("chain", CHAINS)
-def test_rpc_calll(chain):
-    handler = BaseHandler(chain)
-    with pytest.raises(NotImplementedError):
-        handler.rpc_call("foo", {"bar": "baz"})

web3data/handlers/api.py

@@ -1,7 +1,12 @@
 """This module contains the main API handler class."""
 
+from typing import Any, List
+
+import requests
+
 from web3data import __version__
 from web3data.chains import Chains
+from web3data.exceptions import APIError
 from web3data.handlers.address import AddressHandler
 from web3data.handlers.block import BlockHandler
 from web3data.handlers.contract import ContractHandler
@@ -40,7 +45,31 @@ def __init__(self, api_key: str, blockchain_id: str, chain: Chains):
         self.block = BlockHandler(headers, chain)
         self.signature = SignatureHandler(headers, chain)
         self.market = MarketHandler(headers, chain)
-
         self.websocket = WebsocketHandler(
             api_key=self.api_key, blockchain_id=self.blockchain_id
         )
+
+    def rpc(self, method: str, params: List[str], ident: int = 1):
+        """Perform an HTTP POST RPC call on the API.
+
+        Consult the docs here for further details on supported commands:
+        https://docs.amberdata.io/reference#rpc-overview
+
+        :param method: The RPC method to call
+        :param params: Parameters attached to the RPC call
+        :param ident: RPC call identifier
+        """
+        if self.chain not in (Chains.ETH, Chains.ETH_RINKEBY, Chains.BTC):
+            raise APIError(f"RPC calls are not supported for {self.chain}")
+
+        return requests.post(
+            "https://rpc.web3api.io/",
+            json={
+                "jsonrpc": "2.0",
+                "method": method,
+                "params": params,
+                "id": ident,
+            },
+            headers={"x-amberdata-blockchain-id": self.blockchain_id},
+            params={"x-api-key": self.api_key},
+        ).json()

web3data/handlers/base.py

@@ -69,12 +69,3 @@ def raw_query(
             raise EmptyResponseError("The API returned an empty JSON response")
 
         return result
-
-    def rpc_call(self, method: str, params: Dict[str, Any]):
-        """Perform an HTTP POST RPC call on the API.
-
-        :param method: The RPC method to call
-        :param params: Parameters attached to the RPC call
-        """
-        # XXX: always POST at https://rpc.web3api.io/
-        raise NotImplementedError("RPC call integration is not supported yet.")