From 172f00e4e958d7543ccb154956c828937c6f1ef8 Mon Sep 17 00:00:00 2001 From: 2Shirt <2xShirt@gmail.com> Date: Sun, 2 Jul 2023 14:05:20 -0700 Subject: [PATCH] Adjust type hints for NonBlockingStreamReader() --- scripts/wk/exe.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/scripts/wk/exe.py b/scripts/wk/exe.py index dcefc24a..92d48cdd 100644 --- a/scripts/wk/exe.py +++ b/scripts/wk/exe.py @@ -9,7 +9,7 @@ import re import subprocess import time -from io import BufferedReader, TextIOWrapper +from io import IOBase from queue import Queue, Empty from threading import Thread from typing import Any, Callable, Iterable @@ -28,11 +28,11 @@ class NonBlockingStreamReader(): ## https://gist.github.com/EyalAr/7915597 ## https://stackoverflow.com/a/4896288 - def __init__(self, stream: BufferedReader | TextIOWrapper): - self.stream: BufferedReader | TextIOWrapper = stream + def __init__(self, stream: IOBase): + self.stream: IOBase = stream self.queue: Queue = Queue() - def populate_queue(stream: BufferedReader | TextIOWrapper, queue: Queue) -> None: + def populate_queue(stream: IOBase, queue: Queue) -> None: """Collect lines from stream and put them in queue.""" while not stream.closed: try: