(self, command: list, cwd: str)
| 13 | self._stream_tasks = [] # Store stream tasks |
| 14 | |
| 15 | async def run(self, command: list, cwd: str): |
| 16 | if self._is_running or self._is_starting: |
| 17 | await self._output_queue.put({"status": "error", "message": "Process already running"}) |
| 18 | return |
| 19 | try: |
| 20 | self._is_starting = True |
| 21 | # clear the queue before run |
| 22 | while not self._output_queue.empty(): |
| 23 | self._output_queue.get_nowait() |
| 24 | |
| 25 | self._process = await asyncio.create_subprocess_exec( |
| 26 | *command, |
| 27 | cwd=cwd, |
| 28 | stdout=asyncio.subprocess.PIPE, |
| 29 | stderr=asyncio.subprocess.PIPE |
| 30 | ) |
| 31 | self._is_running = True |
| 32 | self._is_starting = False |
| 33 | |
| 34 | async def stream_output(stream, prefix): |
| 35 | while True: |
| 36 | line = await stream.readline() |
| 37 | if line: |
| 38 | message = f"{prefix}{line.decode().strip()}" |
| 39 | print(message,flush=True) #flush output immediately |
| 40 | if prefix == "STDOUT: ": # only add stdout |
| 41 | await self._output_queue.put(message) |
| 42 | |
| 43 | else: |
| 44 | break |
| 45 | |
| 46 | # Create tasks and store them to cancel later |
| 47 | stdout_task = asyncio.create_task(stream_output(self._process.stdout, "STDOUT: ")) |
| 48 | stderr_task = asyncio.create_task(stream_output(self._process.stderr, "STDERR: ")) |
| 49 | self._stream_tasks = [stdout_task, stderr_task] |
| 50 | |
| 51 | # Don't wait, let tasks run |
| 52 | await self._process.wait() |
| 53 | |
| 54 | if self._process.returncode == 0: |
| 55 | await self._output_queue.put({"status": "success", "message": "Process completed successfully"}) |
| 56 | else: |
| 57 | await self._output_queue.put({"status": "error", "message": f"Process exited with code {self._process.returncode}"}) |
| 58 | except Exception as e: |
| 59 | await self._output_queue.put({"status": "error", "message": str(e)}) |
| 60 | finally: |
| 61 | # Cancel the tasks and wait for cancellation to complete |
| 62 | for task in self._stream_tasks: |
| 63 | task.cancel() |
| 64 | try: |
| 65 | await asyncio.gather(*self._stream_tasks, return_exceptions=True) # allow exception |
| 66 | except asyncio.CancelledError: |
| 67 | pass |
| 68 | |
| 69 | self._stream_tasks = [] |
| 70 | self._is_starting = False |
| 71 | self._process = None |
| 72 | self._is_running = False |
no outgoing calls
no test coverage detected