Add mypy types

requirements-dev.txt

@@ -2,6 +2,7 @@ black==19.10b0
 bumpversion==0.5.3
 flake8==3.8.3
 isort==4.3.21
+mypy==0.782
 pytest==5.4.3
 pytest-cov==2.10.0
 semantic_version>=2.8.1,<3

setup.cfg

@@ -7,6 +7,11 @@ current_version = 0.10.1
 max-line-length = 100
 ignore = E203,W503
 
+[mypy]
+disallow_untyped_defs = True
+ignore_missing_imports = True
+follow_imports = silent
+
 [tool:isort]
 force_grid_wrap = 0
 include_trailing_comma = True

solcx/exceptions.py

@@ -9,10 +9,10 @@ class SolcError(Exception):
     def __init__(
         self,
         command: list,
-        return_code: int,
+        return_code: Optional[int],
         stdin_data: Optional[str],
-        stdout_data: str,
-        stderr_data: str,
+        stdout_data: Optional[str],
+        stderr_data: Optional[str],
         message: Optional[str] = None,
     ) -> None:
         if message is not None:

solcx/install.py

@@ -16,7 +16,7 @@
 from base64 import b64encode
 from io import BytesIO
 from pathlib import Path
-from typing import Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 import requests
 from semantic_version import SimpleSpec, Version
@@ -105,7 +105,7 @@ def import_installed_solc(solcx_binary_path: Union[Path, str] = None) -> None:
         .strip()
     )
     if which:
-        path_list.append(Path(which))
+        path_list.append(which)
 
     # on OSX, also copy all versions of solc from cellar
     if platform == "darwin":
@@ -160,7 +160,9 @@ def set_solc_version(
         LOGGER.info(f"Using solc version {solc_version}")
 
 
-def set_solc_version_pragma(pragma_string, silent=False, check_new=False):
+def set_solc_version_pragma(
+    pragma_string: str, silent: bool = False, check_new: bool = False
+) -> None:
     version = _select_pragma_version(pragma_string, get_installed_solc_versions())
     if not version:
         raise SolcNotInstalled(
@@ -247,14 +249,14 @@ def install_solc(
     version: Union[str, Version],
     allow_osx: bool = False,
     show_progress: bool = False,
-    solcx_binary_path: str = None,
+    solcx_binary_path: Union[Path, str] = None,
 ) -> None:
 
     arch = _get_arch()
     platform = _get_platform()
     version = _convert_and_validate_version(version)
 
-    lock = get_process_lock(version)
+    lock = get_process_lock(str(version))
     lock.acquire(True)
 
     try:
@@ -283,7 +285,9 @@ def install_solc(
         lock.release()
 
 
-def _check_subprocess_call(command, message=None, verbose=False, **proc_kwargs):
+def _check_subprocess_call(
+    command: List, message: str = None, verbose: bool = False, **proc_kwargs: Any
+) -> int:
     if message:
         LOGGER.debug(message)
     LOGGER.info(f"Executing: {' '.join(command)}")
@@ -293,7 +297,7 @@ def _check_subprocess_call(command, message=None, verbose=False, **proc_kwargs):
     )
 
 
-def _chmod_plus_x(executable_path):
+def _chmod_plus_x(executable_path: Path) -> None:
     executable_path.chmod(executable_path.stat().st_mode | stat.S_IEXEC)
 
 
@@ -304,15 +308,15 @@ def _check_for_installed_version(
     return path.exists()
 
 
-def _get_temp_folder():
+def _get_temp_folder() -> Path:
     path = Path(tempfile.gettempdir()).joinpath(f"solcx-tmp-{os.getpid()}")
     if path.exists():
         shutil.rmtree(str(path))
     path.mkdir()
     return path
 
 
-def _download_solc(url, show_progress):
+def _download_solc(url: str, show_progress: bool) -> bytes:
     response = requests.get(url, stream=show_progress)
     if response.status_code == 404:
         raise DownloadError(
@@ -321,7 +325,7 @@ def _download_solc(url, show_progress):
         )
     if response.status_code != 200:
         raise DownloadError(
-            f"Received status code {response.status_url} when attempting to download from {url}"
+            f"Received status code {response.status_code} when attempting to download from {url}"
         )
     if not show_progress:
         return response.content
@@ -339,7 +343,7 @@ def _download_solc(url, show_progress):
 
 
 def _install_solc_linux(
-    version: Version, show_progress: bool, solcx_binary_path: Union[Path, str]
+    version: Version, show_progress: bool, solcx_binary_path: Union[Path, str, None]
 ) -> None:
     download = DOWNLOAD_BASE.format(version, "solc-static-linux")
     install_path = get_solc_folder(solcx_binary_path).joinpath(f"solc-v{version}")
@@ -352,7 +356,7 @@ def _install_solc_linux(
 
 
 def _install_solc_windows(
-    version: Version, show_progress: bool, solcx_binary_path: Union[Path, str]
+    version: Version, show_progress: bool, solcx_binary_path: Union[Path, str, None]
 ) -> None:
     download = DOWNLOAD_BASE.format(version, "solidity-windows.zip")
     install_path = get_solc_folder(solcx_binary_path).joinpath(f"solc-v{version}")
@@ -366,13 +370,16 @@ def _install_solc_windows(
 
 
 def _install_solc_arm(
-    version: Version, show_progress: bool, solcx_binary_path: Union[Path, str]
+    version: Version, show_progress: bool, solcx_binary_path: Union[Path, str, None]
 ) -> None:
     _compile_solc(version, show_progress, solcx_binary_path)
 
 
 def _install_solc_osx(
-    version: Version, allow_osx: bool, show_progress: bool, solcx_binary_path: Union[Path, str]
+    version: Version,
+    allow_osx: bool,
+    show_progress: bool,
+    solcx_binary_path: Union[Path, str, None],
 ) -> None:
     if version < Version("0.5.0") and not allow_osx:
         raise ValueError(
@@ -386,7 +393,7 @@ def _install_solc_osx(
 
 
 def _compile_solc(
-    version: Version, show_progress: bool, solcx_binary_path: Union[Path, str]
+    version: Version, show_progress: bool, solcx_binary_path: Union[Path, str, None]
 ) -> None:
     temp_path = _get_temp_folder()
     download = DOWNLOAD_BASE.format(version, f"solidity_{version}.tar.gz")

solcx/main.py

@@ -1,6 +1,6 @@
 import json
 from pathlib import Path
-from typing import Union
+from typing import Dict, List, Union
 
 from semantic_version import Version
 
@@ -20,7 +20,7 @@ def _get_combined_json_outputs() -> str:
     return combined_json_args.split(" ")[-1]
 
 
-def _parse_compiler_output(stdoutdata) -> dict:
+def _parse_compiler_output(stdoutdata: str) -> Dict:
     output = json.loads(stdoutdata)
 
     contracts = output.get("contracts", {})
@@ -38,10 +38,10 @@ def _parse_compiler_output(stdoutdata) -> dict:
 
 def compile_source(
     source: str,
-    output_values: list = None,
-    import_remappings: list = None,
+    output_values: List = None,
+    import_remappings: List = None,
     base_path: str = None,
-    allow_paths: list = None,
+    allow_paths: List = None,
     output_dir: str = None,
     overwrite: bool = False,
     evm_version: str = None,
@@ -55,7 +55,7 @@ def compile_source(
     solc_binary: Union[str, Path] = None,
     solc_version: Version = None,
     allow_empty: bool = False,
-) -> dict:
+) -> Dict:
     if output_values is None:
         combined_json = _get_combined_json_outputs()
     else:
@@ -97,11 +97,11 @@ def compile_source(
 
 
 def compile_files(
-    source_files: list,
-    output_values: list = None,
-    import_remappings: list = None,
+    source_files: List,
+    output_values: List = None,
+    import_remappings: List = None,
     base_path: str = None,
-    allow_paths: list = None,
+    allow_paths: List = None,
     output_dir: str = None,
     overwrite: bool = False,
     evm_version: str = None,
@@ -115,7 +115,7 @@ def compile_files(
     solc_binary: Union[str, Path] = None,
     solc_version: Version = None,
     allow_empty: bool = False,
-) -> dict:
+) -> Dict:
     if output_values is None:
         combined_json = _get_combined_json_outputs()
     else:
@@ -157,18 +157,18 @@ def compile_files(
 
 
 def compile_standard(
-    input_data: dict,
+    input_data: Dict,
     base_path: str = None,
-    allow_paths: list = None,
+    allow_paths: List = None,
     output_dir: str = None,
     overwrite: bool = False,
     solc_binary: Union[str, Path] = None,
     solc_version: Version = None,
     allow_empty: bool = False,
-):
+) -> Dict:
     if not input_data.get("sources") and not allow_empty:
         raise ContractsNotFound(
-            command=None,
+            command=[],
             return_code=None,
             stdin_data=json.dumps(input_data, sort_keys=True, indent=2),
             stdout_data=None,
@@ -210,7 +210,7 @@ def compile_standard(
     return compiler_output
 
 
-def link_code(unlinked_bytecode, libraries):
+def link_code(unlinked_bytecode: str, libraries: Dict) -> str:
     libraries_arg = ",".join(
         (":".join((lib_name, lib_address)) for lib_name, lib_address in libraries.items())
     )

solcx/utils/lock.py

@@ -15,11 +15,11 @@
     NON_BLOCKING = fcntl.LOCK_EX | fcntl.LOCK_NB
     BLOCKING = fcntl.LOCK_EX
 
-_locks: Dict = {}
+_locks: Dict[str, "_ProcessLock"] = {}
 _base_lock = threading.Lock()
 
 
-def get_process_lock(lock_id):
+def get_process_lock(lock_id: str) -> "_ProcessLock":
     with _base_lock:
         if lock_id not in _locks:
             if sys.platform == "win32":
@@ -30,18 +30,24 @@ def get_process_lock(lock_id):
 
 
 class _ProcessLock:
-    def __init__(self, lock_id):
+    def __init__(self, lock_id: str) -> None:
         self._lock = threading.Lock()
         self._lock_path = Path(tempfile.gettempdir()).joinpath(f".solcx-lock-{lock_id}")
         self._lock_file = self._lock_path.open("w")
 
-    def wait(self):
+    def acquire(self, blocking: bool) -> bool:
+        raise NotImplementedError("Method not implemented by child class")
+
+    def release(self) -> None:
+        raise NotImplementedError("Method not implemented by child class")
+
+    def wait(self) -> None:
         self.acquire(True)
         self.release()
 
 
 class UnixLock(_ProcessLock):
-    def acquire(self, blocking):
+    def acquire(self, blocking: bool) -> bool:
         if not self._lock.acquire(blocking):
             return False
         try:
@@ -51,26 +57,28 @@ def acquire(self, blocking):
             return False
         return True
 
-    def release(self):
+    def release(self) -> None:
         fcntl.flock(self._lock_file, fcntl.LOCK_UN)
         self._lock.release()
 
 
 class WindowsLock(_ProcessLock):
-    def acquire(self, blocking):
+    def acquire(self, blocking: bool) -> bool:
         if not self._lock.acquire(blocking):
             return False
         while True:
             try:
-                fd = os.open(self._lock_path, OPEN_MODE)
-                msvcrt.locking(fd, msvcrt.LK_LOCK if blocking else msvcrt.LK_NBLCK, 1)
+                fd = os.open(self._lock_path, OPEN_MODE)  # type: ignore
+                msvcrt.locking(  # type: ignore
+                    fd, msvcrt.LK_LOCK if blocking else msvcrt.LK_NBLCK, 1  # type: ignore
+                )
                 self._fd = fd
                 return True
             except OSError:
                 if not blocking:
                     self._lock.release()
                     return False
 
-    def release(self):
-        msvcrt.locking(self._fd, msvcrt.LK_UNLCK, 1)
+    def release(self) -> None:
+        msvcrt.locking(self._fd, msvcrt.LK_UNLCK, 1)  # type: ignore
         self._lock.release()

solcx/wrapper.py

@@ -1,6 +1,6 @@
 import subprocess
 from pathlib import Path
-from typing import Any, Union
+from typing import Any, List, Tuple, Union
 
 from semantic_version import Version
 
@@ -32,14 +32,14 @@ def solc_wrapper(
     import_remappings: list = None,
     success_return_code: int = None,
     **kwargs: Any,
-):
+) -> Tuple[str, str, list, subprocess.Popen]:
     if solc_binary:
         solc_binary = Path(solc_binary)
     else:
         solc_binary = get_executable()
 
     solc_version = _get_solc_version(solc_binary)
-    command = [solc_binary]
+    command: List = [solc_binary]
 
     if "help" in kwargs:
         success_return_code = 1

tox.ini

@@ -16,3 +16,4 @@ commands =
     black --check {toxinidir}/solcx {toxinidir}/tests
     flake8 {toxinidir}/solcx {toxinidir}/tests
     isort --check-only --diff --recursive {toxinidir}/solcx {toxinidir}/tests
+    mypy {toxinidir}/solcx