|
|
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 |
|
|
|
