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