MCPcopy Create free account
hub / github.com/MaterializeInc/materialize / TaskThread

Class TaskThread

misc/python/materialize/parallel_task.py:41–86  ·  view source on GitHub ↗

Runs a shell command or callable in a thread, capturing output, duration, and exit status.

Source from the content-addressed store, hash-verified

39
40
41class TaskThread(threading.Thread):
42 """Runs a shell command or callable in a thread, capturing output, duration, and exit status."""
43
44 def __init__(
45 self,
46 name: str,
47 spec: TaskSpec | None = None,
48 command: list[str] | None = None,
49 done_queue: queue.Queue[TaskThread] | None = None,
50 ):
51 super().__init__()
52 self.name = name
53 self.output: str = ""
54 self.success = False
55 self.duration: timedelta = timedelta()
56 self._done_queue = done_queue
57 resolved = command if spec is None else spec
58 assert resolved is not None, "must provide spec or command"
59 if callable(resolved):
60 self._fn: Callable[[], tuple[bool, str]] | None = resolved
61 self._command: list[str] | None = None
62 else:
63 self._fn = None
64 self._command = resolved
65
66 def run(self) -> None:
67 start = datetime.now()
68 try:
69 if self._fn is not None:
70 self.success, self.output = self._fn()
71 else:
72 assert self._command is not None
73 proc = subprocess.Popen(
74 self._command,
75 stdout=subprocess.PIPE,
76 stderr=subprocess.STDOUT,
77 )
78 stdout, _ = proc.communicate()
79 self.success = proc.returncode == 0
80 self.output = stdout.decode("utf-8").strip()
81 except Exception as e:
82 self.output = str(e)
83 self.success = False
84 self.duration = datetime.now() - start
85 if self._done_queue is not None:
86 self._done_queue.put(self)
87
88
89class _SpinnerThread(threading.Thread):

Callers 1

run_parallelFunction · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected