Merge pull request #113 from adamzhang1987/feature-new-modules-support

Add two modules and test cases.

Author: corpetty

README.md

@@ -36,6 +36,8 @@ Currently, only the following Etherscan.io API modules are available:
 - proxies
 - blocks
 - transactions
+- Logs
+- Gas Tracker
 
 The remaining available modules provided by Etherscan.io will be added eventually...
 
@@ -58,7 +60,6 @@ Jupyter notebooks area also included in each directory to show all examples
 
 - Package and submit to PyPI
 - Add the following modules:
-  - event logs
   - geth proxy
   - websockets
 - Add robust documentation

etherscan/client.py

@@ -30,6 +30,10 @@ class BadRequest(ClientException):
     """Invalid request passed"""
 
 
+class InvalidAPIKey(ClientException):
+    """Invalid API key"""
+
+
 #  Assume user puts his API key in the api_key.json
 #  file under variable name "key"
 class Client(object):
@@ -59,6 +63,11 @@ class Client(object):
     TAG = '&tag='
     BOOLEAN = '&boolean='
     INDEX = '&index='
+    FROM_BLOCK = '&fromBlock='
+    TO_BLOCK = '&toBlock='
+    TOPIC0 = '&topic0='
+    TOPIC0_1_OPR = '&topic0_1_opr='
+    TOPIC1 = '&topic1='
     API_KEY = '&apikey='
 
     url_dict = {}
@@ -86,7 +95,12 @@ def __init__(self, address, api_key=''):
             (self.TAG, ''),
             (self.BOOLEAN, ''),
             (self.INDEX, ''),
-            (self.API_KEY, api_key)])
+            (self.API_KEY, api_key),
+            (self.FROM_BLOCK, ''),
+            (self.TO_BLOCK, ''),
+            (self.TOPIC0, ''),
+            (self.TOPIC0_1_OPR, ''),
+            (self.TOPIC1, '')])
 
         # Var initialization should take place within init
         self.url = None
@@ -119,6 +133,8 @@ def connect(self):
                 status = data.get('status')
                 if status == '1' or self.check_keys_api(data):
                     return data
+                elif status == '0' and data.get('result') == "Invalid API Key":
+                    raise InvalidAPIKey(data.get('result'))
                 else:
                     raise EmptyResponse(data.get('message', 'no message'))
         raise BadRequest(

etherscan/gas_tracker.py

@@ -0,0 +1,44 @@
+from .client import Client
+
+
+class GasTrackerException(Exception):
+    """Base class for exceptions in this module."""
+    pass
+
+
+class GasTracker(Client):
+    def __init__(self, api_key='YourApiKeyToken'):
+        Client.__init__(self, address='', api_key=api_key)
+        self.url_dict[self.MODULE] = 'gastracker'
+        
+    def get_estimation_of_confirmation_time(self, gas_price: str) -> str:
+        """
+        Returns the estimated time, in seconds, for a transaction to be confirmed on the blockchain.
+
+        Args:
+            gas_price (str): the price paid per unit of gas, in wei
+
+        Returns:
+            str: The result is returned in seconds.
+        """
+        self.url_dict[self.ACTION] = 'gasestimate'
+        self.url_dict[self.GAS_PRICE] = gas_price
+        self.build_url()
+        req = self.connect()
+        return req['result']
+    
+    def get_gas_oracle(self) -> dict:
+        """
+        Returns the current Safe, Proposed and Fast gas prices.
+
+        Returns:
+            dict: The gas prices are returned in Gwei.
+        """
+        self.url_dict[self.ACTION] = 'gasoracle'
+        self.build_url()
+        req = self.connect()
+        return req['result']
+    
+    def get_daily_average_gas_limit(self, start_date, end_date) -> list:
+        # TODO API Pro
+        pass

etherscan/logs.py

@@ -0,0 +1,48 @@
+from .client import Client
+
+
+class LogsException(Exception):
+    """Base class for exceptions in this module."""
+    pass
+
+
+class Logs(Client):
+    """
+    The Event Log API was designed to provide an alternative to the native eth_getLogs.
+    """
+    def __init__(self, api_key='YourApiKeyToken'):
+        Client.__init__(self, address='', api_key=api_key)
+        self.url_dict[self.MODULE] = 'logs'
+    
+    def get_logs(self, from_block: str, to_block='latest', 
+                 topic0='', topic1='', topic0_1_opr='and',) -> list:
+        """
+        Get Event Logs from block number [from_block] to block [to_block] , 
+        where log address = [address], topic[0] = [topic0] 'AND' topic[1] = [topic1]
+
+        Args:
+            from_block (str): start block number
+            to_block (str, optional): end block number. Defaults to 'latest'.
+            topic0 (str, optional): Defaults to ''.
+            topic1 (str, optional): Defaults to ''.
+            topic0_1_opr (str, optional): and|or between topic0 & topic1. Defaults to 'and'.
+
+        Returns:
+            list: [description]
+        """
+        # TODO: support multi topics
+        if not topic0 and topic1:
+            raise(LogsException('can not only set topic1 while topic0 is empty'))
+        self.url_dict[self.ACTION] = 'getLogs'
+        self.url_dict[self.FROM_BLOCK] = from_block if type(
+            from_block) is str else str(from_block)
+        self.url_dict[self.TO_BLOCK] = to_block if type(
+            to_block) is str else str(to_block)
+        self.url_dict[self.TOPIC0] = topic0 if type(
+            topic0) is str else hex(topic0)
+        self.url_dict[self.TOPIC1] = topic1 if type(
+            topic1) is str else hex(topic1)
+        self.url_dict[self.TOPIC0_1_OPR] = topic0_1_opr
+        self.build_url()
+        req = self.connect()
+        return req['result']

setup.py

@@ -2,7 +2,7 @@
 
 setuptools.setup(
     name='py_etherscan_api',
-    version='0.8.0',
+    version='0.9.0',
     packages=['examples', 'examples.stats', 'examples.tokens',
               'examples.accounts', 'examples.blocks', 'examples.transactions',  'etherscan'],
     url='https://github.com/corpetty/py-etherscan-api',

tests/test_accounts.py

@@ -1,24 +1,28 @@
 import unittest
+import warnings
 
 from etherscan.accounts import Account
 
-SINGLE_BALANCE = '40807178566070000000000'
+SINGLE_BALANCE = '40891626854930000000000'
 SINGLE_ACCOUNT = '0xddbd2b932c763ba5b1b7ae3b362eac3e8d40121a'
 MULTI_ACCOUNT = [
     '0xddbd2b932c763ba5b1b7ae3b362eac3e8d40121a',
     '0xddbd2b932c763ba5b1b7ae3b362eac3e8d40121a',
 ]
 MULTI_BALANCE = [
     {'account': '0xddbd2b932c763ba5b1b7ae3b362eac3e8d40121a',
-     'balance': '40807178566070000000000'},
+     'balance': '40891626854930000000000'},
     {'account': '0xddbd2b932c763ba5b1b7ae3b362eac3e8d40121a',
-     'balance': '40807178566070000000000'}
+     'balance': '40891626854930000000000'}
 ]
 API_KEY = 'YourAPIkey'
 
 
 class AccountsTestCase(unittest.TestCase):
 
+    def setUp(self):
+        warnings.simplefilter('ignore', ResourceWarning)
+
     def test_get_balance(self):
         api = Account(address=SINGLE_ACCOUNT, api_key=API_KEY)
         self.assertEqual(api.get_balance(), SINGLE_BALANCE)

tests/test_blocks.py

@@ -1,4 +1,5 @@
 import unittest
+import warnings
 
 from etherscan.blocks import Blocks
 
@@ -10,6 +11,9 @@
 
 class BlocksTestCase(unittest.TestCase):
 
+    def setUp(self):
+        warnings.simplefilter('ignore', ResourceWarning)
+
     def test_get_block_reward(self):
         api = Blocks(api_key=(API_KEY))
         reward_object = api.get_block_reward(BLOCK)

tests/test_gas_tracker.py

@@ -0,0 +1,27 @@
+import unittest
+import warnings
+
+from etherscan.gas_tracker import GasTracker
+
+GAS_PRICE = '2000000000'
+PRICE_ORACLE_RESULT_DICT_KEYS = ("SafeGasPrice",
+                                 "ProposeGasPrice",
+                                 "FastGasPrice",
+                                 "suggestBaseFee")
+API_KEY = 'YourAPIkey'
+
+
+class BlocksTestCase(unittest.TestCase):
+
+    def setUp(self):
+        warnings.simplefilter('ignore', ResourceWarning)
+        self.api = GasTracker(api_key=API_KEY)
+
+    def test_get_estimation_of_confirmation_time(self):
+        estimated_time = self.api.get_estimation_of_confirmation_time(GAS_PRICE)
+        self.assertTrue(int(estimated_time) > 0)
+    
+    def test_get_gas_oracle(self):
+        oracle_price = self.api.get_gas_oracle()
+        for key in PRICE_ORACLE_RESULT_DICT_KEYS:
+            self.assertTrue(key in oracle_price and float(oracle_price[key]) > 0)

tests/test_logs.py

@@ -0,0 +1,40 @@
+import unittest
+import warnings
+
+from etherscan.logs import Logs, LogsException
+from etherscan.client import InvalidAPIKey
+
+FROM_BLOCK = 379224
+TO_BLOCK = 400000
+ADDRESS = '0x33990122638b9132ca29c723bdf037f1a891a70c'
+TOPIC0 = '0xf63780e752c6a54a94fc52715dbc5518a3b4c3c2833d301a204226548a2a8545'
+TOPIC1 = '0x72657075746174696f6e00000000000000000000000000000000000000000000'
+TOPIC0_1_OPR = 'and'
+API_KEY = 'YourAPIkey'
+
+
+class BlocksTestCase(unittest.TestCase):
+
+    def setUp(self):
+        warnings.simplefilter('ignore', ResourceWarning)
+        self.api = Logs(api_key=(API_KEY))
+        
+    def test_invalid_api_key(self):
+        with self.assertRaises(InvalidAPIKey):
+            api = Logs(api_key=('invalid' + API_KEY))
+            api.get_logs(from_block=FROM_BLOCK, topic0=TOPIC0)
+
+    def test_get_logs_error(self):
+        with self.assertRaises(LogsException):
+            self.api.get_logs(from_block=FROM_BLOCK, topic1=TOPIC1)
+
+    def test_get_logs_one_topic(self):
+        topics = self.api.get_logs(from_block=FROM_BLOCK, topic0=TOPIC0)
+        for topic in topics:
+            self.assertTrue(TOPIC0 in topic.get('topics', ''))
+        
+    def test_get_logs_two_topics(self):
+        topics = self.api.get_logs(from_block=FROM_BLOCK, topic0=TOPIC0, topic1=TOPIC1)
+        for topic in topics:
+            self.assertTrue(TOPIC0 in topic.get('topics', '') 
+                            and TOPIC1 in topic.get('topics', ''))

tests/test_proxies.py

@@ -1,5 +1,6 @@
 import re
 import unittest
+import warnings
 
 from etherscan.proxies import Proxies
 
@@ -23,11 +24,14 @@
 
 class ProxiesTestCase(unittest.TestCase):
 
+    def setUp(self):
+        warnings.simplefilter('ignore', ResourceWarning)
+
     def test_get_most_recent_block(self):
         api = Proxies(api_key=API_KEY)
         most_recent = int(api.get_most_recent_block(), 16)
         print(most_recent)
-        p = re.compile('^[0-9]{7}$')
+        p = re.compile('^[0-9]{8}$')
         self.assertTrue(p.match(str(most_recent)))
 
     def test_get_block_by_number(self):