hermes-chatto-plugin/vendor/common/anyio/abc/_subprocesses.py

from __future__ import annotations

from abc import abstractmethod
from signal import Signals
from typing import TYPE_CHECKING

from ._resources import AsyncResource

if TYPE_CHECKING:
    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."""