File size: 2,067 Bytes
4a51346
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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."""