♻️ Refactor console progress printing logic into a dedicated StatusLine class.

concurrent_test/concurrent_test.py

@@ -6,6 +6,23 @@ from time import sleep
 from typing import Any
 
 
+class StatusLine:
+    def __init__(self):
+        self._max_len = 0
+
+    def print(self, line: str):
+        padding_len = self._max_len - len(line)
+        print(f"\r{line}{" " * 0 if padding_len <= 0 else padding_len}", flush=True, end='')
+        self._max_len = max(self._max_len, len(line))
+
+    def clear(self):
+        self.print("")
+        print('\r', flush=True, end='')
+
+    def keep(self):
+        print("", flush=True)
+
+
 def get_item(item: Any) -> str:
     sleep(random.randint(1, 5))
     pass
@@ -17,6 +34,7 @@ def main():
 
     total = len(items)
     count = 0
+    status_line = StatusLine()
     with ThreadPoolExecutor(max_workers=8) as executor:
         futures = []
         for item in items:
@@ -24,9 +42,10 @@ def main():
 
         for future in as_completed(futures):
             count += 1
-            print(f"\r{count}/{total}", flush=True, end="")
+            status_line.print(f"{count}/{total}")
             result.append(future.result())
-        print(f"\r{" " * 10}", flush=True, end="\r")
+        # status_line.clear()
+        status_line.keep()
 
     print(result)