chore: Add mypy to pre-commit

.pre-commit-config.yaml

@@ -24,6 +24,19 @@ repos:
       - id: isort
         # profile and line-length to avoid clashes with black
         args: ["--profile=black", "--line-length=88"]
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.8.0
+    hooks:
+      - id: mypy
+        pass_filenames: false
+        args:
+          - --install-types
+          - --non-interactive
+          - --follow-imports=silent
+          - --ignore-missing-imports
+          - --implicit-optional
+          - --package
+          - boa
 
 default_language_version:
   python: python3.10

Makefile

@@ -4,7 +4,6 @@ all: lint build
 
 lint:
 	pre-commit run --all-files
-	mypy --install-types --non-interactive --follow-imports=silent --ignore-missing-imports --implicit-optional -p boa
 
 build:
 	pip install .

boa/contracts/vyper/ir_executor.py

@@ -365,8 +365,8 @@ def from_mnemonic(cls, mnemonic):
 class OpcodeIRExecutor(IRExecutor):
     _type: type = StackItem  # type: ignore
 
-    def __init__(self, name, opcode_info, *args):
-        self.opcode_info: OpcodeInfo = opcode_info
+    def __init__(self, name: str, opcode_info: OpcodeInfo, *args):
+        self.opcode_info = opcode_info
 
         # to differentiate from implemented codes
         self._name = "__" + name + "__"

boa/contracts/vyper/vyper_contract.py

@@ -14,6 +14,7 @@
 import vyper.ir.compile_ir as compile_ir
 import vyper.semantics.analysis as analysis
 import vyper.semantics.namespace as vy_ns
+from eth.abc import ComputationAPI
 from eth.exceptions import VMError
 from vyper.ast.utils import parse_to_ast
 from vyper.codegen.core import anchor_opt_level, calculate_type_for_external_return
@@ -446,14 +447,14 @@ def __repr__(self):
 class VyperContract(_BaseVyperContract):
     def __init__(
         self,
-        compiler_data,
+        compiler_data: CompilerData,
         *args,
-        env=None,
-        override_address=None,
+        env: Env = None,
+        override_address: Address = None,
         # whether to skip constructor
         skip_initcode=False,
-        created_from=None,
-        filename=None,
+        created_from: Address = None,
+        filename: str = None,
     ):
         super().__init__(compiler_data, env, filename)
 
@@ -471,11 +472,11 @@ def __init__(
         if "__init__" in external_fns:
             self._ctor = VyperFunction(external_fns.pop("__init__"), self)
 
-        if skip_initcode:
-            addr = Address(override_address)
-        else:
-            addr = self._run_init(*args, override_address=override_address)
-        self._address: Address = addr
+        self._address: Address = (
+            Address(override_address)
+            if skip_initcode
+            else self._run_init(*args, override_address=override_address)
+        )
 
         for fn_name, fn in external_fns.items():
             setattr(self, fn_name, VyperFunction(fn, self))
@@ -490,12 +491,13 @@ def __init__(
         self._storage = StorageModel(self)
 
         self._eval_cache = lrudict(0x1000)
-        self._source_map = None
-        self._computation = None
+        self._source_map: dict | None = None
+        self._computation: ComputationAPI | None = None
+        self.bytecode: bytes | None = None
 
         self.env.register_contract(self._address, self)
 
-    def _run_init(self, *args, override_address=None):
+    def _run_init(self, *args, override_address=None) -> Address:
         encoded_args = b""
         if self._ctor:
             encoded_args = self._ctor.prepare_calldata(*args)

boa/environment.py

@@ -6,7 +6,7 @@
 import random
 import sys
 import warnings
-from typing import Any, Iterator, Optional, Tuple
+from typing import Any, Iterator, Optional, Tuple, TypeAlias
 
 import eth.constants as constants
 import eth.tools.builder.chain as chain
@@ -96,7 +96,7 @@ def anchor(self):
 
 
 # make mypy happy
-_AddressType = Address | str | bytes | PYEVM_Address
+_AddressType: TypeAlias = Address | str | bytes | PYEVM_Address
 
 
 _opcode_overrides = {}

boa/explorer.py

@@ -1,12 +1,11 @@
 import json
-from typing import Optional
 
 import requests
 
 SESSION = requests.Session()
 
 
-def _fetch_etherscan(uri: str, api_key: Optional[str] = None, **params) -> dict:
+def _fetch_etherscan(uri: str, api_key: str | None, **params) -> dict:
     if api_key is not None:
         params["apikey"] = api_key
 
@@ -35,13 +34,12 @@ def fetch_abi_from_etherscan(
 
 # fetch the address of a contract; resolves at most one layer of indirection
 # if the address is a proxy contract.
-def _resolve_implementation_address(address: str, uri: str, api_key: Optional[str]):
+def _resolve_implementation_address(address: str, uri: str, api_key: str | None):
     params = dict(module="contract", action="getsourcecode", address=address)
     data = _fetch_etherscan(uri, api_key, **params)
     source_data = data["result"][0]
 
     # check if the contract is a proxy
     if int(source_data["Proxy"]) == 1:
         return source_data["Implementation"]
-    else:
-        return address
+    return address

boa/integrations/jupyter/__init__.py

@@ -16,10 +16,10 @@ def load_jupyter_server_extension(server_app):
 _load_jupyter_server_extension = load_jupyter_server_extension
 
 
-__all__ = [  # type: ignore
-    BrowserSigner,
-    BrowserRPC,
-    BrowserEnv,
-    load_jupyter_server_extension,
-    _load_jupyter_server_extension,
+__all__ = [
+    "BrowserSigner",
+    "BrowserRPC",
+    "BrowserEnv",
+    "load_jupyter_server_extension",
+    "_load_jupyter_server_extension",
 ]

boa/interpret.py

@@ -125,4 +125,4 @@ def from_etherscan(
     return ABIContractFactory.from_abi_dict(abi, name=name).at(addr)
 
 
-__all__ = []  # type: ignore
+__all__: list[str] = []

boa/network.py

@@ -4,7 +4,6 @@
 from dataclasses import dataclass
 from functools import cached_property
 from math import ceil
-from typing import Any
 
 from eth_account import Account
 from requests.exceptions import HTTPError
@@ -189,9 +188,12 @@ def get_eip1559_fee(self) -> tuple[str, str, str, str]:
         max_fee = to_hex(base_fee_estimate + to_int(max_priority_fee))
         return to_hex(base_fee_estimate), max_priority_fee, max_fee, chain_id
 
-    def get_static_fee(self) -> list[Any]:
+    def get_static_fee(self) -> tuple[str, str]:
         # non eip-1559 transaction
-        return self._rpc.fetch_multi([("eth_gasPrice", []), ("eth_chainId", [])])
+        price, chain_id = self._rpc.fetch_multi(
+            [("eth_gasPrice", []), ("eth_chainId", [])]
+        )
+        return price, chain_id
 
     def _check_sender(self, address):
         if address is None:
@@ -371,7 +373,7 @@ def _reset_fork(self, block_identifier="latest"):
             block_identifier=block_identifier,
             cache_file=None,
         )
-        self.vm.state._account_db._rpc._init_mem_db()
+        self.vm.state._account_db._caching_rpc._init_mem_db()
 
     def _send_txn(self, from_, to=None, gas=None, value=None, data=None):
         tx_data = fixup_dict(

boa/rpc.py

@@ -47,9 +47,9 @@ def to_bytes(hex_str: str) -> bytes:
 
 
 class RPCError(Exception):
-    def __init__(self, message, code):
+    def __init__(self, message: str, code: int):
         super().__init__(f"{code}: {message}")
-        self.code: str = code
+        self.code = code
 
     @classmethod
     def from_json(cls, data):
@@ -111,21 +111,21 @@ def name(self):
         parse_result = urlparse(self._rpc_url)
         return f"{parse_result.scheme}://{parse_result.netloc} (URL partially masked for privacy)"
 
-    def fetch(self, method, params):
+    def fetch(self, method, params) -> Any:
         # the obvious thing to do here is dispatch into fetch_multi.
         # but some providers (alchemy) can't handle batched requests
         # for certain endpoints (debug_traceTransaction).
-        req = {"jsonrpc": "2.0", "method": method, "params": params, "id": 0}
+        request = {"jsonrpc": "2.0", "method": method, "params": params, "id": 0}
         # print(req)
-        res = self._session.post(self._rpc_url, json=req, timeout=TIMEOUT)
-        res.raise_for_status()
-        res = json.loads(res.text)
+        response = self._session.post(self._rpc_url, json=request, timeout=TIMEOUT)
+        response.raise_for_status()
+        response_body: dict = json.loads(response.text)
         # print(res)
-        if "error" in res:
-            raise RPCError.from_json(res["error"])
-        return res["result"]
+        if "error" in response_body:
+            raise RPCError.from_json(response_body["error"])
+        return response_body["result"]
 
-    def fetch_multi(self, payloads):
+    def fetch_multi(self, payloads) -> list:
         request = [
             {"jsonrpc": "2.0", "method": method, "params": params, "id": i}
             for i, (method, params) in enumerate(payloads)

boa/vm/fork.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any
+from typing import Any, Dict
 
 from requests import HTTPError
 
@@ -120,21 +120,22 @@ def fetch_multi(self, payload):
 # AccountDB which dispatches to an RPC when we don't have the
 # data locally
 class AccountDBFork(AccountDB):
-    _rpc: RPC = None  # type: ignore
-    _rpc_init_kwargs: dict[str, Any] = {}
+    _rpc: RPC | None = None
+    _rpc_init_kwargs: Dict[str, Any] = {}
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
 
         rpc_kwargs = self._rpc_init_kwargs.copy()
 
         block_identifier = rpc_kwargs.pop("block_identifier", "safe")
-        self._rpc: CachingRPC = CachingRPC(self._rpc, **rpc_kwargs)
+        assert self._rpc, "RPC not initialized"
+        self._caching_rpc: CachingRPC = CachingRPC(self._rpc, **rpc_kwargs)
 
         if block_identifier not in _PREDEFINED_BLOCKS:
             block_identifier = to_hex(block_identifier)
 
-        self._block_info = self._rpc.fetch_uncached(
+        self._block_info = self._caching_rpc.fetch_uncached(
             "eth_getBlockByNumber", [block_identifier, False]
         )
         self._block_number = to_int(self._block_info["number"])
@@ -178,7 +179,7 @@ def _get_account_rpc(self, address):
             ("eth_getTransactionCount", [addr, self._block_id]),
             ("eth_getCode", [addr, self._block_id]),
         ]
-        res = self._rpc.fetch_multi(reqs)
+        res = self._caching_rpc.fetch_multi(reqs)
         balance = to_int(res[0])
         nonce = to_int(res[1])
         code = to_bytes(res[2])
@@ -201,7 +202,7 @@ def try_prefetch_state(self, msg: Message):
         # arguments with this specific block before
         try:
             tracer = {"tracer": "prestateTracer"}
-            res = self._rpc.fetch_uncached(
+            res = self._caching_rpc.fetch_uncached(
                 "debug_traceCall", [args, self._block_id, tracer]
             )
         except (RPCError, HTTPError):
@@ -245,7 +246,7 @@ def get_code(self, address):
         try:
             return super().get_code(address)
         except MissingBytecode:  # will get thrown if code_hash != hash(empty)
-            ret = self._rpc.fetch(
+            ret = self._caching_rpc.fetch(
                 "eth_getCode", [to_checksum_address(address), self._block_id]
             )
             return to_bytes(ret)
@@ -269,7 +270,9 @@ def get_storage(self, address, slot, from_journal=True):
             return s
 
         addr = to_checksum_address(address)
-        ret = self._rpc.fetch("eth_getStorageAt", [addr, to_hex(slot), self._block_id])
+        ret = self._caching_rpc.fetch(
+            "eth_getStorageAt", [addr, to_hex(slot), self._block_id]
+        )
         return to_int(ret)
 
     def account_exists(self, address):