| | from __future__ import annotations |
| |
|
| | from abc import abstractmethod |
| | from signal import Signals |
| |
|
| | from ._resources import AsyncResource |
| | from ._streams import ByteReceiveStream, ByteSendStream |
| |
|
| |
|
| | class Process(AsyncResource): |
| | """An asynchronous version of :class:`subprocess.Popen`.""" |
| |
|
| | @abstractmethod |
| | async def wait(self) -> int: |
| | """ |
| | Wait until the process exits. |
| | |
| | :return: the exit code of the process |
| | """ |
| |
|
| | @abstractmethod |
| | def terminate(self) -> None: |
| | """ |
| | Terminates the process, gracefully if possible. |
| | |
| | On Windows, this calls ``TerminateProcess()``. |
| | On POSIX systems, this sends ``SIGTERM`` to the process. |
| | |
| | .. seealso:: :meth:`subprocess.Popen.terminate` |
| | """ |
| |
|
| | @abstractmethod |
| | def kill(self) -> None: |
| | """ |
| | Kills the process. |
| | |
| | On Windows, this calls ``TerminateProcess()``. |
| | On POSIX systems, this sends ``SIGKILL`` to the process. |
| | |
| | .. seealso:: :meth:`subprocess.Popen.kill` |
| | """ |
| |
|
| | @abstractmethod |
| | def send_signal(self, signal: Signals) -> None: |
| | """ |
| | Send a signal to the subprocess. |
| | |
| | .. seealso:: :meth:`subprocess.Popen.send_signal` |
| | |
| | :param signal: the signal number (e.g. :data:`signal.SIGHUP`) |
| | """ |
| |
|
| | @property |
| | @abstractmethod |
| | def pid(self) -> int: |
| | """The process ID of the process.""" |
| |
|
| | @property |
| | @abstractmethod |
| | def returncode(self) -> int | None: |
| | """ |
| | The return code of the process. If the process has not yet terminated, this will |
| | be ``None``. |
| | """ |
| |
|
| | @property |
| | @abstractmethod |
| | def stdin(self) -> ByteSendStream | None: |
| | """The stream for the standard input of the process.""" |
| |
|
| | @property |
| | @abstractmethod |
| | def stdout(self) -> ByteReceiveStream | None: |
| | """The stream for the standard output of the process.""" |
| |
|
| | @property |
| | @abstractmethod |
| | def stderr(self) -> ByteReceiveStream | None: |
| | """The stream for the standard error output of the process.""" |
| |
|