Merge pull request #581 from enzbang/mypy-cleanup

Mypy cleanup

Author: enzbang

.pre-commit-config.yaml

@@ -24,3 +24,4 @@ repos:
         - types-PyYAML
         - types-requests
         - types-setuptools
+        - types-psutil

src/e3/anod/sandbox/__init__.py

@@ -147,7 +147,7 @@ def get_configuration(self) -> dict:
             return yaml.safe_load(f)
 
     def write_scripts(self) -> None:
-        from setuptools.command.easy_install import get_script_args  # type: ignore[attr-defined]
+        from setuptools.command.easy_install import ScriptWriter
 
         # Retrieve sandbox_scripts entry points
         e3_distrib = get_distribution("e3-core")
@@ -161,7 +161,7 @@ def get_entry_map(self, group):  # type: ignore
             def as_requirement(self):  # type: ignore
                 return e3_distrib.as_requirement()
 
-        for script in get_script_args(dist=SandboxDist()):
+        for script in ScriptWriter.best().get_args(dist=SandboxDist()):
             script_name = script[0]
             script_content = script[1]
             target = os.path.join(self.bin_dir, script_name)
@@ -171,8 +171,8 @@ def as_requirement(self):  # type: ignore
                     "console_scripts", "sandbox_scripts"
                 )
             with open(target, "wb") as f:
-                if isinstance(script_content, str):
-                    f.write(script_content.encode("utf-8"))
+                if isinstance(script_content, bytes):  # type: ignore[unreachable]
+                    f.write(script_content)  # type: ignore[unreachable]
                 else:
-                    f.write(script_content)
+                    f.write(script_content.encode("utf-8"))
             chmod("a+x", target)

src/e3/net/http.py

@@ -11,8 +11,8 @@
 import requests
 import requests.adapters
 import requests.exceptions
-import requests.packages.urllib3.exceptions
-from requests.packages.urllib3.util import Retry
+import urllib3.exceptions
+from urllib3.util import Retry
 
 from typing import TYPE_CHECKING
 
@@ -221,7 +221,7 @@ def request(
             except (
                 socket.timeout,
                 requests.exceptions.RequestException,
-                requests.packages.urllib3.exceptions.HTTPError,
+                urllib3.exceptions.HTTPError,
             ) as e:
                 # got an error with that base url so put it last in our list
                 error_msgs.append(f"{message_prefix}{e}")

src/e3/os/process.py

@@ -56,10 +56,12 @@
 try:
     import psutil
     from psutil import Popen
+
+    has_psutil = True
 except ImportError:  # defensive code
-    from subprocess import Popen
+    from subprocess import Popen  # type: ignore
 
-    psutil = None
+    has_psutil = False
 
 
 def subprocess_setup() -> None:
@@ -419,7 +421,7 @@ def add_interpreter_command(cmd_line: CmdLine) -> CmdLine:
                 self.internal = Popen(self.cmds[0], **popen_args)
 
             else:
-                runs: list[subprocess.Popen] = []
+                runs: list[Popen] = []
                 for index, cmd in enumerate(self.cmds):
                     if index == 0:
                         stdin: int | IO[Any] = self.input_file.fd
@@ -609,15 +611,15 @@ def interrupt(self) -> None:
 
     def is_running(self) -> bool:
         """Check whether the process is running."""
-        if psutil is None:  # defensive code
+        if not has_psutil:  # defensive code
             # psutil not imported, use our is_running function
             return is_running(self.pid)
         else:
             return self.internal.is_running()
 
     def children(self) -> list[Any]:
         """Return list of child processes (using psutil)."""
-        if psutil is None:  # defensive code
+        if not has_psutil:  # defensive code
             raise NotImplementedError("Run.children() require psutil")
         return self.internal.children()
 
@@ -838,7 +840,7 @@ def kill_process_tree(pid: int | Any, timeout: int = 3) -> bool:
         except psutil.NoSuchProcess:  # defensive code
             pass
 
-    def on_terminate(p: str) -> None:
+    def on_terminate(p: psutil.Process) -> None:
         """Log info when a process terminate."""
         logger.debug("process %s killed", p)