chore: refactor (#171)

* chore: refactor

* chore: fix args and kwargs type hinting for Contract

dank_mids/brownie_patch/_abi.py

@@ -0,0 +1,23 @@
+
+import functools
+from typing import Any
+
+from brownie.convert.utils import build_function_selector, build_function_signature
+from web3.datastructures import AttributeDict
+
+
+@functools.lru_cache(maxsize=None)
+class FunctionABI:
+    """A singleton to hold function signatures, since we only really need one copy of each func sig in memory."""
+    __slots__ = "abi", "input_sig", "signature"
+    def __init__(self, **abi: Any):
+        self.abi = abi
+        self.input_sig = build_function_signature(abi)
+        self.signature = build_function_selector(abi)
+
+def _make_hashable(obj: Any) -> Any:
+    if isinstance(obj, (list, tuple)):
+        return tuple((_make_hashable(o) for o in obj))
+    elif isinstance(obj, dict):
+        return AttributeDict({k: _make_hashable(v) for k, v in obj.items()})
+    return obj

dank_mids/brownie_patch/_method.py

@@ -0,0 +1,107 @@
+
+import asyncio
+import functools
+from decimal import Decimal
+from typing import Any, Awaitable, Callable, Dict, Generic, Iterable, List, Optional, TypeVar
+
+from brownie.typing import AccountsType
+from brownie.convert.datatypes import EthAddress
+from eth_abi.exceptions import InsufficientDataBytes
+from web3 import Web3
+
+from dank_mids import ENVIRONMENT_VARIABLES as ENVS
+from dank_mids.brownie_patch._abi import FunctionABI, _make_hashable
+
+_EVMType = TypeVar("_EVMType")
+
+class _DankMethodMixin(Generic[_EVMType]):
+    _address: EthAddress
+    _abi: FunctionABI
+    """A mixin class that gives brownie objects async support and reduces memory usage"""
+    __slots__ = "_address", "_abi", "_name", "_owner", "natspec", "_encode_input", "_decode_input"
+    def __await__(self):
+        """Asynchronously call the contract method without arguments at the latest block and await the result."""
+        return self.coroutine().__await__()
+    async def map(
+        self, 
+        args: Iterable[Any], 
+        block_identifier: Optional[int] = None,
+        decimals: Optional[int] = None,
+    ) -> List[_EVMType]:
+        return await asyncio.gather(*[self.coroutine(arg, block_identifier=block_identifier, decimals=decimals) for arg in args])
+    @property
+    def abi(self) -> dict:
+        return self._abi.abi
+    @property
+    def signature(self) -> str:
+        return self._abi.signature
+    @property
+    def _input_sig(self) -> str:
+        return self._abi.input_sig
+    @functools.cached_property
+    def _len_inputs(self) -> int:
+        return len(self.abi['inputs'])
+    @functools.cached_property
+    def _skip_decoder_proc_pool(self) -> bool:
+        from dank_mids.brownie_patch.call import _skip_proc_pool
+        return self._address in _skip_proc_pool
+    @functools.cached_property
+    def _web3(cls) -> Web3:
+        from dank_mids import web3
+        return web3
+    @functools.cached_property
+    def _prep_request_data(self) -> Callable[..., Awaitable[bytes]]:
+        from dank_mids.brownie_patch import call
+        if ENVS.OPERATION_MODE.application or self._len_inputs:
+            return call.encode
+        else:
+            return call._request_data_no_args
+
+class _DankMethod(_DankMethodMixin):
+    __slots__ = "_address", "_abi", "_name", "_owner", "natspec", "_encode_input", "_decode_output"
+    def __init__(
+        self,
+        address: str,
+        abi: Dict,
+        name: str,
+        owner: Optional[AccountsType],
+        natspec: Optional[Dict] = None,
+    ) -> None:
+        self._address = address
+        self._abi = FunctionABI(**{key: _make_hashable(abi[key]) for key in sorted(abi)})
+        self._name = name
+        self._owner = owner
+        self.natspec = natspec or {}
+        # TODO: refactor this
+        from dank_mids.brownie_patch import call
+        self._encode_input = call.encode_input
+        self._decode_output = call.decode_output
+    async def coroutine(  # type: ignore [empty-body]
+        self, 
+        *args: Any, 
+        block_identifier: Optional[int] = None, 
+        decimals: Optional[int] = None,
+        override: Optional[Dict[str, str]] = None,
+    ) -> _EVMType:
+        """
+        Asynchronously call the contract method via dank mids and await the result.
+        
+        Arguments:
+            - *args: The arguments for the contract method.
+            - block_identifier (optional): The block at which the chain will be read. If not provided, will read the chain at latest block.
+            - decimals (optional): if provided, the output will be `result / 10 ** decimals`
+        
+        Returns:
+            - Whatever the node sends back as the output for this contract method.
+        """
+        if override:
+            raise ValueError("Cannot use state override with `coroutine`.")
+        async with ENVS.BROWNIE_ENCODER_SEMAPHORE[block_identifier]:
+            data = await self._encode_input(self, self._len_inputs, self._prep_request_data, *args)
+            async with ENVS.BROWNIE_CALL_SEMAPHORE[block_identifier]:
+                output = await self._web3.eth.call({"to": self._address, "data": data}, block_identifier)
+        try:
+            decoded = await self._decode_output(self, output)
+        except InsufficientDataBytes as e:
+            raise InsufficientDataBytes(str(e), self, self._address, output) from e
+        return decoded if decimals is None else decoded / 10 ** Decimal(decimals)

dank_mids/brownie_patch/contract.py

@@ -1,6 +1,6 @@
 
 import functools
-from typing import Dict, List, NewType, Optional, Union, overload
+from typing import Dict, List, Literal, NewType, Optional, Union, overload
 
 import brownie
 from brownie.network.contract import (ContractCall, ContractTx, OverloadedMethod, 
@@ -21,14 +21,38 @@
 class Contract(brownie.Contract):
     """a modified `brownie.Contract` with async and call batching functionalities"""
     @classmethod
-    def from_abi(cls, *args, **kwargs) -> "Contract":
-        return Contract(brownie.Contract.from_abi(*args, **kwargs).address)
+    def from_abi(
+        cls, 
+        name: str, 
+        address: str, 
+        abi: List[dict], 
+        owner: Optional[AccountsType] = None, 
+        persist: bool = True,
+    ) -> "Contract":
+        persisted = brownie.Contract.from_abi(name, address, abi, owner, _check_persist(persist))
+        return Contract(persisted.address)
     @classmethod
-    def from_ethpm(cls, *args, **kwargs) -> "Contract":
-        return Contract(brownie.Contract.from_ethpm(*args, **kwargs).address)
+    def from_ethpm(
+        cls, 
+        name: str, 
+        manifest_uri: str, 
+        address: Optional[str] = None, 
+        owner: Optional[AccountsType] = None, 
+        persist: bool = True,
+    ) -> "Contract":
+        persisted = brownie.Contract.from_ethpm(name, manifest_uri, address, owner, _check_persist(persist))
+        return Contract(persisted.address)
     @classmethod
-    def from_explorer(cls, *args, **kwargs) -> "Contract":
-        return Contract(brownie.Contract.from_explorer(*args, **kwargs).address)
+    def from_explorer(
+        cls, 
+        address: str, 
+        as_proxy_for: Optional[str] = None, 
+        owner: Optional[AccountsType] = None, 
+        silent: bool = False, 
+        persist: bool = True,
+    ) -> "Contract":
+        persisted = brownie.Contract.from_explorer(address, as_proxy_for, owner, silent, _check_persist(persist))
+        return Contract(persisted.address)
     topics: Dict[str, str]
     signatures: Dict[Method, Signature]
     def __init__(self, *args, **kwargs):
@@ -96,3 +120,8 @@ def _patch_if_method(method: ContractMethod, w3: Web3) -> None:
 
 class _ContractMethodPlaceholder:
     """A sentinel object that indicates a Contract does have a member by a specific name."""
+
+def _check_persist(persist: bool) -> Literal[True]:
+    if persist is False:
+        raise NotImplementedError("persist: False")
+    return persist

dank_mids/brownie_patch/types.py

@@ -1,132 +1,14 @@
 
-import asyncio
-import functools
-from decimal import Decimal
-from typing import Any, Awaitable, Callable, Dict, Generic, Iterable, List, Optional, Tuple, TypeVar, Union
+from typing import Any, Dict, Optional, Tuple, Union
 
 from brownie.typing import AccountsType
-from brownie.convert.datatypes import EthAddress
-from brownie.convert.utils import build_function_selector, build_function_signature
 from brownie.network.contract import ContractCall, ContractTx, OverloadedMethod
-from eth_abi.exceptions import InsufficientDataBytes
-from web3 import Web3
-from web3.datastructures import AttributeDict
 
-from dank_mids import ENVIRONMENT_VARIABLES as ENVS
+from dank_mids.brownie_patch._method import _DankMethod, _DankMethodMixin, _EVMType
 
-_EVMType = TypeVar("_EVMType")
 
 ContractMethod = Union[ContractCall, ContractTx, OverloadedMethod]
 
-@functools.lru_cache(maxsize=None)
-class FunctionABI:
-    """A singleton to hold function signatures, since we only really need one copy of each func sig in memory."""
-    __slots__ = "abi", "input_sig", "signature"
-    def __init__(self, **abi: Any):
-        self.abi = abi
-        self.input_sig = build_function_signature(abi)
-        self.signature = build_function_selector(abi)
-
-
-def _make_hashable(obj: Any) -> Any:
-    if isinstance(obj, (list, tuple)):
-        return tuple((_make_hashable(o) for o in obj))
-    elif isinstance(obj, dict):
-        return AttributeDict({k: _make_hashable(v) for k, v in obj.items()})
-    return obj
-
-class _DankMethodMixin(Generic[_EVMType]):
-    _address: EthAddress
-    _abi: FunctionABI
-    """A mixin class that gives brownie objects async support and reduces memory usage"""
-    __slots__ = "_address", "_abi", "_name", "_owner", "natspec", "_encode_input", "_decode_input"
-    def __await__(self):
-        """Asynchronously call the contract method without arguments at the latest block and await the result."""
-        return self.coroutine().__await__()
-    async def map(
-        self, 
-        args: Iterable[Any], 
-        block_identifier: Optional[int] = None,
-        decimals: Optional[int] = None,
-    ) -> List[_EVMType]:
-        return await asyncio.gather(*[self.coroutine(arg, block_identifier=block_identifier, decimals=decimals) for arg in args])
-    @property
-    def abi(self) -> dict:
-        return self._abi.abi
-    @property
-    def signature(self) -> str:
-        return self._abi.signature
-    @property
-    def _input_sig(self) -> str:
-        return self._abi.input_sig
-    @functools.cached_property
-    def _len_inputs(self) -> int:
-        return len(self.abi['inputs'])
-    @functools.cached_property
-    def _skip_decoder_proc_pool(self) -> bool:
-        from dank_mids.brownie_patch.call import _skip_proc_pool
-        return self._address in _skip_proc_pool
-    @functools.cached_property
-    def _web3(cls) -> Web3:
-        from dank_mids import web3
-        return web3
-    @functools.cached_property
-    def _prep_request_data(self) -> Callable[..., Awaitable[bytes]]:
-        from dank_mids.brownie_patch import call
-        if ENVS.OPERATION_MODE.application or self._len_inputs:
-            return call.encode
-        else:
-            return call._request_data_no_args
-
-class _DankMethod(_DankMethodMixin):
-    __slots__ = "_address", "_abi", "_name", "_owner", "natspec", "_encode_input", "_decode_output"
-    def __init__(
-        self,
-        address: str,
-        abi: Dict,
-        name: str,
-        owner: Optional[AccountsType],
-        natspec: Optional[Dict] = None,
-    ) -> None:
-        self._address = address
-        self._abi = FunctionABI(**{key: _make_hashable(abi[key]) for key in sorted(abi)})
-        self._name = name
-        self._owner = owner
-        self.natspec = natspec or {}
-        # TODO: refactor this
-        from dank_mids.brownie_patch import call
-        self._encode_input = call.encode_input
-        self._decode_output = call.decode_output
-    async def coroutine(  # type: ignore [empty-body]
-        self, 
-        *args: Any, 
-        block_identifier: Optional[int] = None, 
-        decimals: Optional[int] = None,
-        override: Optional[Dict[str, str]] = None,
-    ) -> _EVMType:
-        """
-        Asynchronously call the contract method via dank mids and await the result.
-        
-        Arguments:
-            - *args: The arguments for the contract method.
-            - block_identifier (optional): The block at which the chain will be read. If not provided, will read the chain at latest block.
-            - decimals (optional): if provided, the output will be `result / 10 ** decimals`
-        
-        Returns:
-            - Whatever the node sends back as the output for this contract method.
-        """
-        if override:
-            raise ValueError("Cannot use state override with `coroutine`.")
-        async with ENVS.BROWNIE_ENCODER_SEMAPHORE[block_identifier]:
-            data = await self._encode_input(self, self._len_inputs, self._prep_request_data, *args)
-            async with ENVS.BROWNIE_CALL_SEMAPHORE[block_identifier]:
-                output = await self._web3.eth.call({"to": self._address, "data": data}, block_identifier)
-        try:
-            decoded = await self._decode_output(self, output)
-        except InsufficientDataBytes as e:
-            raise InsufficientDataBytes(str(e), self, self._address, output) from e
-        return decoded if decimals is None else decoded / 10 ** Decimal(decimals)
-
 class DankContractCall(_DankMethod, ContractCall):
     """
     A `brownie.network.contract.ContractCall` subclass with async support via the `coroutine` method.