Skip to content
Snippets Groups Projects

Add lots of type hints, fix some bugs, and clean up some code

Merged Newbyte requested to merge newbyte/more-mypy into master
1 file
+ 33
8
Compare changes
  • Side-by-side
  • Inline
  • I think the "these types make no sense" comment were about
    output_return_buffer, which defaulted to False but was to be set to
    list[bytes] if used. I changed it so that it defaults to None, which
    probably is more conventional. Other than that I didn't notice anything
    weird about the types here.
    
    Also check for None as necessary to appease mypy.
+ 33
8
@@ -14,6 +14,7 @@ import sys
import threading
import time
from collections.abc import Sequence
from typing import overload, Literal
import pmb.helpers.run
"""For a detailed description of all output modes, read the description of
@@ -94,14 +95,33 @@ def pipe(cmd, working_dir=None):
return ret
# FIXME (#2324): These types make no sense at all
@overload
def pipe_read(
process,
output_to_stdout=False,
output_log=True,
output_return=False,
output_return_buffer=False,
):
process: subprocess.Popen,
output_to_stdout: bool = ...,
output_log: bool = ...,
output_return: Literal[False] = ...,
output_return_buffer: None = ...,
) -> None: ...
@overload
def pipe_read(
process: subprocess.Popen,
output_to_stdout: bool = ...,
output_log: bool = ...,
output_return: Literal[True] = ...,
output_return_buffer: list[bytes] = ...,
) -> None: ...
def pipe_read(
process: subprocess.Popen,
output_to_stdout: bool = False,
output_log: bool = True,
output_return: bool = False,
output_return_buffer: list[bytes] | None = None,
) -> None:
"""Read all output from a subprocess, copy it to the log and optionally stdout and a buffer variable.
This is only meant to be called by foreground_pipe() below.
@@ -115,13 +135,18 @@ def pipe_read(
"""
while True:
# Copy available output
out = process.stdout.readline()
process_stdout = process.stdout
if process_stdout is None:
raise RuntimeError("subprocess has no stdout?")
out = process_stdout.readline()
if len(out):
if output_log:
pmb.helpers.logging.logfd.buffer.write(out)
if output_to_stdout:
sys.stdout.buffer.write(out)
if output_return:
if output_return_buffer is None:
raise AssertionError
output_return_buffer.append(out)
continue
Loading