Refactored structure

Author: Kotsias, Panagiotis-Christos

etherscan/modules/__init__.py

@@ -0,0 +1,265 @@
+from functools import reduce
+from typing import List
+
+from etherscan.utils.datatypes import TxHash, WalletAddress
+from etherscan.enums.actions_enum import ActionsEnum as actions
+from etherscan.enums.fields_enum import FieldsEnum as fields
+from etherscan.enums.modules_enum import ModulesEnum as modules
+from etherscan.enums.tags_enum import TagsEnum as tags
+
+
+class Accounts:
+    @staticmethod
+    def get_eth_balance(wallet: WalletAddress) -> str:
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.BALANCE}"
+            f"{fields.ADDRESS}"
+            f"{wallet}"
+            f"{fields.TAG}"
+            f"{tags.LATEST}"
+        )
+        return url
+        # r = requests.get(url)
+        # return conversions.to_ticker_unit(parser.get_result(r))
+
+    @staticmethod
+    def get_eth_balance_multiple(wallets: List[WalletAddress]) -> str:
+        # max 20 wallets
+        wallet_list = reduce(lambda w1, w2: str(w1) + "," + str(w2), wallets)
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.BALANCE_MULTI}"
+            f"{fields.ADDRESS}"
+            f"{wallet_list}"
+            f"{fields.TAG}"
+            f"{tags.LATEST}"
+        )
+        return url
+        # r = requests.get(url)
+        # return [conversions.to_ticker_unit(r["balance"]) for r in parser.get_result(r)]
+
+    @staticmethod
+    def get_hist_eth_balance_by_block(wallet: WalletAddress, block: int) -> str:
+        # throttled to 2 calls/sec
+        # BUG: returns 'Error! Missing Or invalid Action name'
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.BALANCE_HISTORY}"
+            f"{fields.ADDRESS}"
+            f"{wallet}"
+            f"{fields.BLOCKNO}"
+            f"{str(block)}"
+        )
+        return url
+
+    @staticmethod
+    def get_normal_txs_by_address(
+        wallet: WalletAddress,
+        startblock: int = 0000,
+        endblock: int = 99999999,
+        sort: str = "asc",
+    ) -> str:
+        # last 10,000 txs only
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.TXLIST}"
+            f"{fields.ADDRESS}"
+            f"{wallet}"
+            f"{fields.START_BLOCK}"
+            f"{str(startblock)}"
+            f"{fields.END_BLOCK}"
+            f"{str(endblock)}"
+            f"{fields.SORT}"
+            f"{sort}"
+        )
+        return url
+
+    @staticmethod
+    def get_normal_txs_by_address_paginated(
+        wallet: WalletAddress,
+        page: int,
+        offset: int,
+        startblock: int = 0000,
+        endblock: int = 99999999,
+        sort: str = "asc",
+    ) -> str:
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.TXLIST}"
+            f"{fields.ADDRESS}"
+            f"{wallet}"
+            f"{fields.START_BLOCK}"
+            f"{str(startblock)}"
+            f"{fields.END_BLOCK}"
+            f"{str(endblock)}"
+            f"{fields.SORT}"
+            f"{sort}"
+            f"{fields.PAGE}"
+            f"{str(page)}"
+            f"{fields.OFFSET}"
+            f"{str(offset)}"
+        )
+        return url
+
+    @staticmethod
+    def get_internal_txs_by_address(
+        wallet: WalletAddress,
+        startblock: int = 0000,
+        endblock: int = 99999999,
+        sort: str = "asc",
+    ) -> str:
+        # last 10,000 txs only
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.TXLIST_INTERNAL}"
+            f"{fields.ADDRESS}"
+            f"{wallet}"
+            f"{fields.START_BLOCK}"
+            f"{str(startblock)}"
+            f"{fields.END_BLOCK}"
+            f"{str(endblock)}"
+            f"{fields.SORT}"
+            f"{sort}"
+        )
+        return url
+
+    @staticmethod
+    def get_internal_txs_by_address_paginated(
+        wallet: WalletAddress,
+        page: int,
+        offset: int,
+        startblock: int = 0000,
+        endblock: int = 99999999,
+        sort: str = "asc",
+    ) -> str:
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.TXLIST_INTERNAL}"
+            f"{fields.ADDRESS}"
+            f"{wallet}"
+            f"{fields.START_BLOCK}"
+            f"{str(startblock)}"
+            f"{fields.END_BLOCK}"
+            f"{str(endblock)}"
+            f"{fields.SORT}"
+            f"{sort}"
+            f"{fields.PAGE}"
+            f"{str(page)}"
+            f"{fields.OFFSET}"
+            f"{str(offset)}"
+        )
+        return url
+
+    @staticmethod
+    def get_internal_txs_by_txhash(txhash: TxHash) -> str:
+        # last 10,000 txs only
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.TXLIST_INTERNAL}"
+            f"{fields.TXHASH}"
+            f"{txhash}"
+        )
+        return url
+
+    @staticmethod
+    def get_internal_txs_by_block_range_paginated(
+        startblock: int, endblock: int, page: int, offset: int, sort: str = "asc",
+    ) -> str:
+        # last 10,000 txs only
+        # BUG: returns empty message
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.TXLIST_INTERNAL}"
+            f"{fields.START_BLOCK}"
+            f"{str(startblock)}"
+            f"{fields.END_BLOCK}"
+            f"{str(endblock)}"
+            f"{fields.SORT}"
+            f"{sort}"
+            f"{fields.PAGE}"
+            f"{str(page)}"
+            f"{fields.OFFSET}"
+            f"{str(offset)}"
+        )
+        return url
+
+    @staticmethod
+    def get_erc20_token_transfer_events_by_address(
+        wallet: WalletAddress,
+        startblock: int = 0,
+        endblock: int = 999999999,
+        sort: str = "asc",
+    ) -> str:
+        # last 10,000 txs only
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.TOKENTX}"
+            f"{fields.ADDRESS}"
+            f"{wallet}"
+            f"{fields.START_BLOCK}"
+            f"{str(startblock)}"
+            f"{fields.END_BLOCK}"
+            f"{str(endblock)}"
+            f"{fields.SORT}"
+            f"{sort}"
+        )
+        return url
+
+    @staticmethod
+    def get_erc721_token_transfer_events_by_address(
+        wallet: WalletAddress,
+        startblock: int = 0,
+        endblock: int = 999999999,
+        sort: str = "asc",
+    ) -> str:
+        # last 10,000 txs only
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.TOKENNFTTX}"
+            f"{fields.ADDRESS}"
+            f"{wallet}"
+            f"{fields.START_BLOCK}"
+            f"{str(startblock)}"
+            f"{fields.END_BLOCK}"
+            f"{str(endblock)}"
+            f"{fields.SORT}"
+            f"{sort}"
+        )
+        return url
+
+    @staticmethod
+    def get_mined_blocks_by_address(wallet: WalletAddress) -> str:
+        url = (
+            f"{fields.MODULE}"
+            f"{modules.ACCOUNT}"
+            f"{fields.ACTION}"
+            f"{actions.GET_MINED_BLOCKS}"
+            f"{fields.ADDRESS}"
+            f"{wallet}"
+            f"{fields.BLOCK_TYPE}"
+            f"blocks"
+        )
+        return url

etherscan/modules/accounts.py

@@ -0,0 +1,2 @@
+class Contracts:
+    pass

etherscan/modules/contracts.py

@@ -0,0 +1,2 @@
+class Tokens:
+    pass

etherscan/modules/tokens.py

@@ -0,0 +1,2 @@
+class Transactions:
+    pass

etherscan/modules/transactions.py

@@ -0,0 +1,13 @@
+from decimal import Decimal
+
+
+class Conversions:
+    @staticmethod
+    def to_ticker_unit(val: int, decimals: int = 18) -> Decimal:
+        factor = Decimal("10") ** Decimal("-{}".format(decimals))
+        return Decimal(val) * factor
+
+    @staticmethod
+    def to_smallest_unit(val: int, decimals: int = 18) -> Decimal:
+        factor = Decimal("10") ** Decimal("+{}".format(decimals))
+        return Decimal(val) * factor

etherscan/utils/__init__.py

@@ -0,0 +1,52 @@
+from etherscan.enums.messages_enum import MessagesEnum as msgs
+
+_CONTRACT_LEN = 42
+_TX_LEN = 66
+_WALLET_LEN = 42
+
+
+class BaseAddress:
+    def __init__(self):
+        self._address = ""
+        self._maxlen = 0
+
+    def _check_0x(self, address: str):
+        if address[:2] != "0x":
+            raise ValueError(msgs.INVALID_ADDRESS)
+        return address
+
+    def _check_len(self, address: str):
+        if len(address) != self._maxlen:
+            raise ValueError(msgs.INVALID_ADDRESS)
+        return address
+
+    def __repr__(self):
+        return self._address
+
+    def __str__(self):
+        return self._address
+
+    def __add__(self, val: str):
+        return self._address + val
+
+    def __len__(self):
+        return len(self._address)
+
+
+class ContractAddress(BaseAddress):
+    def __init__(self, address: str):
+        self._maxlen = _CONTRACT_LEN
+        self._address = self._check_len(self._check_0x(address))
+
+
+class TxHash(BaseAddress):
+    def __init__(self, address: str):
+        self._maxlen = _TX_LEN
+        self._address = self._check_len(self._check_0x(address))
+
+
+class WalletAddress(BaseAddress):
+    def __init__(self, address: str):
+        self._maxlen = _WALLET_LEN
+        self._address = self._check_len(self._check_0x(address))
+

etherscan/utils/conversions.py

@@ -0,0 +1,22 @@
+import requests
+
+
+class ResponseParser:
+    @staticmethod
+    def _get_content(response: requests.Response):
+        return response.json()
+
+    @staticmethod
+    def get_status(response: requests.Response):
+        c = ResponseParser._get_content(response)
+        return c["status"]
+
+    @staticmethod
+    def get_message(response: requests.Response):
+        c = ResponseParser._get_content(response)
+        return c["message"]
+
+    @staticmethod
+    def get_result(response: requests.Response):
+        c = ResponseParser._get_content(response)
+        return c["result"]