Merge pull request #99 from iamdefinitelyahuman/feat-mypy

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