mypy fixes

working pre-commit/dmypy config

.pre-commit-config.yaml

@@ -65,6 +65,7 @@ repos:
         entry: dmypy
         files: \.py$
         language: python
+        require_serial: true
         args: ["run", "--", "--strict", "--implicit-reexport", "--warn-unused-ignores", "--cache-fine-grained"]
 
       - id: pylint

bombard/pretty_sz.py

@@ -1,4 +1,7 @@
-def pretty_sz(size: int) -> str:
+from typing import Union
+
+
+def pretty_sz(size: Union[int, float]) -> str:
     dividers = {
         "bytes": 1,
         "kb": 1024,
@@ -12,3 +15,4 @@ def pretty_sz(size: int) -> str:
         if result < 100:
             return f"{result:.1f} {unit}"
         result = round(result)
+    return f"{result:.1f} {dividers['pb']}"

bombard/show_descr.py

@@ -4,4 +4,4 @@
 
 
 def markdown_for_terminal(descr: str) -> str:
-    return highlight(descr, MarkdownLexer(), TerminalFormatter())
+    return str(highlight(descr, MarkdownLexer(), TerminalFormatter()))

bombard/weaver_mill.py

@@ -9,6 +9,7 @@
 from copy import deepcopy
 from queue import Queue
 from threading import Thread
+from typing import Any, Dict
 
 
 class WeaverMill:
@@ -18,15 +19,15 @@ def __init__(self, threads_num: int = 10):
         """
         self.threads_num = threads_num
         self.threads = []
-        self.queue = Queue()
+        self.queue: Queue = Queue()
         self.job_count = 0
         for thread_id in range(threads_num):
             t = Thread(target=self.thread_worker, args=[thread_id])
             t.daemon = True
             t.start()
             self.threads.append(t)
 
-    def thread_worker(self, thread_id):
+    def thread_worker(self, thread_id: int) -> None:
         """
         Get job from queue and pass it to abstract worker
         that should be implemented in descendant.
@@ -45,26 +46,26 @@ def thread_worker(self, thread_id):
                 self.queue.task_done()
 
     @abstractmethod
-    def worker(self, thread_id, job):
+    def worker(self, thread_id: int, job: Dict[str, Any]) -> None:
         """
         Implement your job processor, runs in thread.
 
         :param thread_id: just sequential number of the thread we work into
         :param job: job from queue
         """
 
-    def put(self, job):
+    def put(self, job: Dict[str, Any]) -> None:
         """
         Add job to queue.
         To start processing use `process`.
         """
         self.queue.put(job)
 
-    def process(self):
+    def process(self) -> None:
         """ Starts all threads and lock until queue is empty """
         self.queue.join()
 
-    def stop(self):
+    def stop(self) -> None:
         """ Stops all threads - send stop signal to queue and lock until they stop """
         for _ in self.threads:
             self.queue.put(None)

lint.sh

@@ -1,3 +1,3 @@
-dmypy start -- --use-fine-grained-cache
+#dmypy start -- --use-fine-grained-cache
 #dmypy run --verbose -- --strict --implicit-reexport --warn-unused-ignores --cache-fine-grained .
 (time pre-commit run --verbose --all-files) 2>&1

requirements.txt

@@ -1,3 +1,4 @@
 pyyaml>=5.1
 pygments
 colorama
+types-PyYAML