| | from asyncio import create_subprocess_exec, subprocess |
| |
|
| |
|
| | class Terminal: |
| | """ |
| | Class for running terminal commands asynchronously. |
| | |
| | Methods: |
| | |
| | run(commands: str) |
| | commands: Terminal Commands. |
| | Returns Process id (int) |
| | |
| | terminate(pid: int) |
| | pid: Process id returned in `run` method. |
| | Returns True if terminated else False (bool) |
| | |
| | output(pid: int) |
| | pid: Process id returned in `run` method. |
| | Returns Output of process (str) |
| | |
| | error(pid: int) |
| | pid: Process id returned in `run` method. |
| | Returns Error of process (str) |
| | """ |
| |
|
| | def __init__(self) -> None: |
| | self._processes = {} |
| |
|
| | @staticmethod |
| | def _to_str(data: bytes) -> str: |
| | return data.decode("utf-8").strip() |
| |
|
| | async def run(self, *args) -> int: |
| | process = await create_subprocess_exec( |
| | *args, stdout=subprocess.PIPE, stderr=subprocess.PIPE |
| | ) |
| | pid = process.pid |
| | self._processes[pid] = process |
| | return pid |
| |
|
| | def terminate(self, pid: int) -> bool: |
| | try: |
| | self._processes.pop(pid) |
| | self._processes[pid].kill() |
| | return True |
| | except KeyError: |
| | return False |
| |
|
| | async def output(self, pid: int) -> str: |
| | output = [] |
| | while True: |
| | out = self._to_str(await self._processes[pid].stdout.readline()) |
| | if not out: |
| | break |
| | output.append(out) |
| | return "\n".join(output) |
| |
|
| | async def error(self, pid: int) -> str: |
| | error = [] |
| | while True: |
| | err = self._to_str(await self._processes[pid].stderr.readline()) |
| | if not err: |
| | break |
| | error.append(err) |
| | return "\n".join(error) |
| |
|
| | @property |
| | def _auto_remove_processes(self) -> None: |
| | while self._processes: |
| | for proc in self._processes.keys(): |
| | if proc.returncode is not None: |
| | try: |
| | self._processes.pop(proc) |
| | except KeyError: |
| | pass |
| |
|
