1from __future__ import annotations
2
3from abc import abstractmethod
4from signal import Signals
5from typing import TYPE_CHECKING
6
7from ._resources import AsyncResource
8
9if TYPE_CHECKING:
10 from ._streams import ByteReceiveStream, ByteSendStream
11
12
13class Process(AsyncResource):
14 """An asynchronous version of :class:`subprocess.Popen`."""
15
16 @abstractmethod
17 async def wait(self) -> int:
18 """
19 Wait until the process exits.
20
21 :return: the exit code of the process
22 """
23
24 @abstractmethod
25 def terminate(self) -> None:
26 """
27 Terminates the process, gracefully if possible.
28
29 On Windows, this calls ``TerminateProcess()``.
30 On POSIX systems, this sends ``SIGTERM`` to the process.
31
32 .. seealso:: :meth:`subprocess.Popen.terminate`
33 """
34
35 @abstractmethod
36 def kill(self) -> None:
37 """
38 Kills the process.
39
40 On Windows, this calls ``TerminateProcess()``.
41 On POSIX systems, this sends ``SIGKILL`` to the process.
42
43 .. seealso:: :meth:`subprocess.Popen.kill`
44 """
45
46 @abstractmethod
47 def send_signal(self, signal: Signals) -> None:
48 """
49 Send a signal to the subprocess.
50
51 .. seealso:: :meth:`subprocess.Popen.send_signal`
52
53 :param signal: the signal number (e.g. :data:`signal.SIGHUP`)
54 """
55
56 @property
57 @abstractmethod
58 def pid(self) -> int:
59 """The process ID of the process."""
60
61 @property
62 @abstractmethod
63 def returncode(self) -> int | None:
64 """
65 The return code of the process. If the process has not yet terminated, this will
66 be ``None``.
67 """
68
69 @property
70 @abstractmethod
71 def stdin(self) -> ByteSendStream | None:
72 """The stream for the standard input of the process."""
73
74 @property
75 @abstractmethod
76 def stdout(self) -> ByteReceiveStream | None:
77 """The stream for the standard output of the process."""
78
79 @property
80 @abstractmethod
81 def stderr(self) -> ByteReceiveStream | None:
82 """The stream for the standard error output of the process."""