style: disallow untyped defs, add missing types

setup.cfg

@@ -8,6 +8,7 @@ max-line-length = 100
 ignore = E203,W503
 
 [mypy]
+disallow_untyped_defs = True
 ignore_missing_imports = True
 follow_imports = silent
 

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
@@ -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(
@@ -254,7 +256,7 @@ def install_solc(
     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

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,15 +157,15 @@ 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=[],
@@ -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, List, Union
+from typing import Any, List, Tuple, Union
 
 from semantic_version import Version
 
@@ -32,7 +32,7 @@ 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: