|
56 | 56 | try:
|
57 | 57 | import psutil
|
58 | 58 | from psutil import Popen
|
| 59 | + |
| 60 | + has_psutil = True |
59 | 61 | except ImportError: # defensive code
|
60 |
| - from subprocess import Popen |
| 62 | + from subprocess import Popen # type: ignore |
61 | 63 |
|
62 |
| - psutil = None |
| 64 | + has_psutil = False |
63 | 65 |
|
64 | 66 |
|
65 | 67 | def subprocess_setup() -> None:
|
@@ -419,7 +421,7 @@ def add_interpreter_command(cmd_line: CmdLine) -> CmdLine:
|
419 | 421 | self.internal = Popen(self.cmds[0], **popen_args)
|
420 | 422 |
|
421 | 423 | else:
|
422 |
| - runs: list[subprocess.Popen] = [] |
| 424 | + runs: list[Popen] = [] |
423 | 425 | for index, cmd in enumerate(self.cmds):
|
424 | 426 | if index == 0:
|
425 | 427 | stdin: int | IO[Any] = self.input_file.fd
|
@@ -609,15 +611,15 @@ def interrupt(self) -> None:
|
609 | 611 |
|
610 | 612 | def is_running(self) -> bool:
|
611 | 613 | """Check whether the process is running."""
|
612 |
| - if psutil is None: # defensive code |
| 614 | + if not has_psutil: # defensive code |
613 | 615 | # psutil not imported, use our is_running function
|
614 | 616 | return is_running(self.pid)
|
615 | 617 | else:
|
616 | 618 | return self.internal.is_running()
|
617 | 619 |
|
618 | 620 | def children(self) -> list[Any]:
|
619 | 621 | """Return list of child processes (using psutil)."""
|
620 |
| - if psutil is None: # defensive code |
| 622 | + if not has_psutil: # defensive code |
621 | 623 | raise NotImplementedError("Run.children() require psutil")
|
622 | 624 | return self.internal.children()
|
623 | 625 |
|
@@ -838,7 +840,7 @@ def kill_process_tree(pid: int | Any, timeout: int = 3) -> bool:
|
838 | 840 | except psutil.NoSuchProcess: # defensive code
|
839 | 841 | pass
|
840 | 842 |
|
841 |
| - def on_terminate(p: str) -> None: |
| 843 | + def on_terminate(p: psutil.Process) -> None: |
842 | 844 | """Log info when a process terminate."""
|
843 | 845 | logger.debug("process %s killed", p)
|
844 | 846 |
|
|
0 commit comments