Skip to content

API Reference

process

Process

A class for spawning, managing, and interacting with a process synchronously.

This class provides a convenient interface for running a process, interacting with its standard input, output, and error streams, and managing its execution.

Source code in process/process.py 
 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
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
class Process(AbstractProcess[StreamWriter, StreamReader]):
    """A class for spawning, managing, and interacting with a process synchronously.

    This class provides a convenient interface for running a process, interacting with its standard input, output, and error streams, and managing its execution.
    """

    def __init__(
        self,
        arguments: Union[StrOrPath, Sequence[StrOrPath]],
        stdin: Optional[Union[bytes, File]] = None,
        stdout: Optional[File] = PIPE,
        stderr: Optional[File] = PIPE,
        buffer_size: Optional[int] = None,
    ) -> None:
        """Initialize a new [`Process`][process.Process] instance with the given `arguments`.

        Args:
            arguments: The command and its arguments for the process.
            stdin: A [`bytes`][bytes] object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard input.
            stdout: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard output.
            stderr: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], [`DEVNULL`][process.DEVNULL], and [`STDOUT`][process.STDOUT]) to use as the standard error.
            buffer_size: The buffer size for the stream operations. If `None`, the default buffer size will be used. If `0`, all operations will be unbuffered.

        Notes:
            If `arguments` is an instance of [`str`][str], it will be split into a list of arguments using [`shlex.split()`][shlex.split].

        Warning:
            A file-like object without a valid file descriptor, such as [`BytesIO`][io.BytesIO], cannot be used as `stdin`, `stdout`, or `stderr`.
        """
        if buffer_size is None:
            buffer_size = DEFAULT_BUFFER_SIZE

        super().__init__(arguments, stdin=stdin, stdout=stdout, stderr=stderr, buffer_size=buffer_size)

    def __del__(self) -> None:
        """Clean up resources used by the process.

        This method cleans up resources as follows:

        - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
        - If the process is running, terminate the process.
        - Wait for the process to complete.
        - Close the standard output stream if the process was created with `stdout=PIPE` and the stream is not closed.
        - Close the standard error stream if the process was created with `stderr=PIPE` and the stream is not closed.
        """
        if self._process is None:
            return

        with suppress(ProcessInvalidStreamError):
            # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
            # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
            with suppress(BrokenPipeError, ConnectionResetError):
                if not self.stdin.closed:
                    self.stdin.close()

        if self.running:
            self.terminate()
            self.join()

        with suppress(ProcessInvalidStreamError):
            if not self.stdout.closed:
                self.stdout.close()

        with suppress(ProcessInvalidStreamError):
            if not self.stderr.closed:
                self.stderr.close()

    if sys.version_info >= (3, 9):

        @property
        def process(self) -> subprocess.Popen[bytes]:
            """Return the underlying process instance.

            Returns:
                An instance of [`subprocess.Popen`][subprocess.Popen].

            Raises:
                ProcessNotRunError: If the process has not been run.
            """
            if self._process is None:
                raise ProcessNotRunError("Process has not been run")

            return cast(subprocess.Popen[bytes], self._process)
    else:

        @property
        def process(self) -> subprocess.Popen:
            """Return the underlying process instance.

            Returns:
                An instance of [`subprocess.Popen`][subprocess.Popen].

            Raises:
                ProcessNotRunError: If the process has not been run.
            """
            if self._process is None:
                raise ProcessNotRunError("Process has not been run")

            return cast(subprocess.Popen, self._process)

    @property
    def running(self) -> bool:
        """Check if the process is currently running.

        Returns:
            `True` if the process is running, `False` otherwise.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        return self.process.poll() is None

    def run(self) -> Self:
        """Run the process.

        Returns:
            The current [`Process`][process.Process] instance itself.

        Raises:
            ProcessAlreadyRunError: If the process has already been run.
            ProcessError: If the process fails to run.
        """
        if self._process is not None:
            raise ProcessAlreadyRunError("Process has already been run")

        creationflags = 0

        # According to the documentation for [`Popen.send_signal()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/subprocess.rst?plain=1#L864),
        # > On Windows, ... CTRL_C_EVENT and CTRL_BREAK_EVENT can be sent to processes
        # > started with a *creationflags* parameter which includes ``CREATE_NEW_PROCESS_GROUP``.
        if is_windows():
            creationflags = subprocess.CREATE_NEW_PROCESS_GROUP  # type: ignore

        stdin = self._stdin
        should_feed_stdin = isinstance(stdin, bytes)

        try:
            self._process = subprocess.Popen(
                self.arguments,
                bufsize=self._buffer_size,
                stdin=cast(File, stdin) if not should_feed_stdin else PIPE,
                stdout=self._stdout,
                stderr=self._stderr,
                creationflags=creationflags,
            )
        except Exception as exception:
            raise ProcessError(f"Failed to run process: {exception!r}")

        if should_feed_stdin:
            # Since `Popen.communicate()` requires the process to terminate, we feed the input ourselves.
            # If deadlock issues are reported, consider using a different approach, such as threading.
            self.stdin.write(cast(bytes, stdin))
            self.stdin.close()

        return self

    def output(self, join: bool = True) -> bytes:
        r"""Return the output from the standard output stream of the process if the process was created with `stdout=PIPE`.

        Warning:
            If the standard output stream of the process is modified by others, the result of [`output()`][process.Process.output] will be affected by those changes.

            ```python
            with Process("echo something") as process:
                print(process.stdout.read(4))  # b'some'
            print(process.output())  # b'thing\n'
            ```

        Args:
            join: Whether to wait for the process to complete before returning the output.

        Returns:
            The output from the standard output stream of the process if the process was created with `stdout=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
        """
        if join:
            self.join()

        if not self.stdout.closed:
            self._output += self.stdout.read()

        return self._output

    def error(self, join: bool = True) -> bytes:
        """Return the output from the standard error stream of the process if the process was created with `stderr=PIPE`.

        Warning:
            If the standard error stream of the process is modified by others, the result of [`error()`][process.Process.error] will be affected by those changes.

        Args:
            join: Whether to wait for the process to complete before returning the output.

        Returns:
            The output from the standard error stream of the process if the process was created with `stderr=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
        """
        if join:
            self.join()

        if not self.stderr.closed:
            self._error += self.stderr.read()

        return self._error

    def join(self, timeout: Optional[float] = None) -> None:
        """Wait for the process to complete.

        Args:
            timeout: Maximum time to wait for the process to complete, in seconds. If `None`, wait indefinitely.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessTimeoutError: If the process does not complete within the specified timeout.
        """
        try:
            self.process.wait(timeout)
        except subprocess.TimeoutExpired:
            raise ProcessTimeoutError(f"Process did not complete within {timeout} seconds")

    def close(self) -> None:
        """Close the process and release its resources.

        This method cleans up the process execution as follows:

        - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
        - If the process is running, terminate it.
        - Wait for the process to complete.
        - Read the remaining output from the standard output stream and then close the stream if the process was created with `stdout=PIPE` and the stream is not closed.
        - Read the remaining error from the standard error stream and then close the stream if the process was created with `stderr=PIPE` and the stream is not closed.
        """
        with suppress(ProcessInvalidStreamError):
            # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
            # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
            with suppress(BrokenPipeError, ConnectionResetError):
                if not self.stdin.closed:
                    self.stdin.close()

        self.terminate()
        self.join()

        with suppress(ProcessInvalidStreamError):
            if not self.stdout.closed:
                self._output += self.stdout.read()
                self.stdout.close()

        with suppress(ProcessInvalidStreamError):
            if not self.stderr.closed:
                self._error += self.stderr.read()
                self.stderr.close()

    def __enter__(self) -> Self:
        """Enter the runtime context for this [`Process`][process.Process] instance.

        This method runs the process.

        Returns:
            The current [`Process`][process.Process] instance itself.
        """
        return self.run()

    def __exit__(
        self,
        exception_type: Optional[type[BaseException]],
        exception: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> None:
        """Exit the runtime context for this [`Process`][process.Process] instance.

        This method cleans up the process execution as follows:

        - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
        - Wait for the process to complete.
        - Read the remaining output from the standard output stream and then close the stream if the process was created with `stdout=PIPE` and the stream is not closed.
        - Read the remaining error from the standard error stream and then close the stream if the process was created with `stderr=PIPE` and the stream is not closed.
        """
        with suppress(ProcessInvalidStreamError):
            # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
            # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
            with suppress(BrokenPipeError, ConnectionResetError):
                if not self.stdin.closed:
                    self.stdin.close()

        if exception_type is None:
            self.join()

        with suppress(ProcessInvalidStreamError):
            if not self.stdout.closed:
                if exception_type is None:
                    self._output += self.stdout.read()

                self.stdout.close()

        with suppress(ProcessInvalidStreamError):
            if not self.stderr.closed:
                if exception_type is None:
                    self._error += self.stderr.read()

                self.stderr.close()

    @property
    def stdin(self) -> StreamWriter:
        """Return the standard input stream of the process if the process was created with `stdin=PIPE`.

        Returns:
            The standard input stream of the process if the process was created with `stdin=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdin=PIPE`.
        """
        stdin = self.process.stdin

        if stdin is None:
            raise ProcessInvalidStreamError("Process was not created with `stdin=PIPE`")

        # According to the documentation for [`Popen.stdin`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/subprocess.rst?plain=1#L892)
        # > If the *stdin* argument was :data:`PIPE`, this attribute is a writeable stream object as returned by :func:`open`.
        # According to the documentation for [`open`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/functions.rst?plain=1#L1418)
        # > When used to open a file in a binary mode with buffering, the returned class is a subclass of :class:`io.BufferedIOBase`.
        # > The exact class varies: in read binary mode, it returns an :class:`io.BufferedReader`;
        # > in write binary and append binary modes, it returns an :class:`io.BufferedWriter`, and
        # > in read/write mode, it returns an :class:`io.BufferedRandom`.
        # > When buffering is disabled, the raw stream, a subclass of :class:`io.RawIOBase`, :class:`io.FileIO`, is returned.
        if isinstance(stdin, BufferedIOBase):
            return cast(BufferedWriter, stdin)
        elif isinstance(stdin, RawIOBase):
            return cast(FileIO, stdin)
        else:
            assert False, "Unreachable code"

    @property
    def stdout(self) -> StreamReader:
        """Return the standard output stream of the process if the process was created with `stdout=PIPE`.

        Returns:
            The standard output stream of the process if the process was created with `stdout=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
        """
        stdout = self.process.stdout

        if stdout is None:
            raise ProcessInvalidStreamError("Process was not created with `stdout=PIPE`")

        # According to the documentation for [`Popen.stdout`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/subprocess.rst?plain=1#L901)
        # > If the *stdout* argument was :data:`PIPE`, this attribute is a readable stream object as returned by :func:`open`.
        # According to the documentation for [`open`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/functions.rst?plain=1#L1418)
        # > When used to open a file in a binary mode with buffering, the returned class is a subclass of :class:`io.BufferedIOBase`.
        # > The exact class varies: in read binary mode, it returns an :class:`io.BufferedReader`;
        # > in write binary and append binary modes, it returns an :class:`io.BufferedWriter`, and
        # > in read/write mode, it returns an :class:`io.BufferedRandom`.
        # > When buffering is disabled, the raw stream, a subclass of :class:`io.RawIOBase`, :class:`io.FileIO`, is returned.
        if isinstance(stdout, BufferedIOBase):
            return cast(BufferedReader, stdout)
        elif isinstance(stdout, RawIOBase):
            return cast(FileIO, stdout)
        else:
            assert False, "Unreachable code"

    @property
    def stderr(self) -> StreamReader:
        """Return the standard error stream of the process if the process was created with `stderr=PIPE`.

        Returns:
            The standard error stream of the process if the process was created with `stderr=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
        """
        stderr = self.process.stderr

        if stderr is None:
            raise ProcessInvalidStreamError("Process was not created with `stderr=PIPE`")

        # According to the documentation for [`Popen.stderr`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/subprocess.rst?plain=1#L911)
        # > If the *stderr* argument was :data:`PIPE`, this attribute is a readable stream object as returned by :func:`open`.
        # According to the documentation for [`open`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/functions.rst?plain=1#L1418)
        # > When used to open a file in a binary mode with buffering, the returned class is a subclass of :class:`io.BufferedIOBase`.
        # > The exact class varies: in read binary mode, it returns an :class:`io.BufferedReader`;
        # > in write binary and append binary modes, it returns an :class:`io.BufferedWriter`, and
        # > in read/write mode, it returns an :class:`io.BufferedRandom`.
        # > When buffering is disabled, the raw stream, a subclass of :class:`io.RawIOBase`, :class:`io.FileIO`, is returned.
        if isinstance(stderr, BufferedIOBase):
            return cast(BufferedReader, stderr)
        elif isinstance(stderr, RawIOBase):
            return cast(FileIO, stderr)
        else:
            assert False, "Unreachable code"

process property 

process: Popen

Return the underlying process instance.

Returns:

Type Description
Popen

An instance of subprocess.Popen.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

running property 

running: bool

Check if the process is currently running.

Returns:

Type Description
bool

True if the process is running, False otherwise.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

stdin property 

stdin: StreamWriter

Return the standard input stream of the process if the process was created with stdin=PIPE.

Returns:

Type Description
StreamWriter

The standard input stream of the process if the process was created with stdin=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdin=PIPE.

stdout property 

stdout: StreamReader

Return the standard output stream of the process if the process was created with stdout=PIPE.

Returns:

Type Description
StreamReader

The standard output stream of the process if the process was created with stdout=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdout=PIPE.

stderr property 

stderr: StreamReader

Return the standard error stream of the process if the process was created with stderr=PIPE.

Returns:

Type Description
StreamReader

The standard error stream of the process if the process was created with stderr=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stderr=PIPE.

arguments property 

arguments: list[str]

Return the command-line arguments provided to create this Process instance.

Returns:

Type Description
list[str]

The command-line arguments provided to create this Process instance.

id property 

id: int

Return the process identifier.

Returns:

Type Description
int

The process identifier.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

exit_code property 

exit_code: int | None

Return the exit code of the process.

Returns:

Type Description
int | None

The exit code of the process. If the process is still running, it returns None.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

__init__ 

__init__(
    arguments: StrOrPath | Sequence[StrOrPath],
    stdin: bytes | File | None = None,
    stdout: File | None = PIPE,
    stderr: File | None = PIPE,
    buffer_size: int | None = None,
) -> None

Initialize a new Process instance with the given arguments.

Parameters:

Name Type Description Default
arguments StrOrPath | Sequence[StrOrPath]

The command and its arguments for the process.

 required
stdin bytes | File | None

A bytes object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, and DEVNULL) to use as the standard input.

 None
stdout File | None

An existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, and DEVNULL) to use as the standard output.

 PIPE
stderr File | None

An existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, DEVNULL, and STDOUT) to use as the standard error.

 PIPE
buffer_size int | None

The buffer size for the stream operations. If None, the default buffer size will be used. If 0, all operations will be unbuffered.

 None
Notes

If arguments is an instance of str, it will be split into a list of arguments using shlex.split().

Warning

A file-like object without a valid file descriptor, such as BytesIO, cannot be used as stdin, stdout, or stderr.

Source code in process/process.py 
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
def __init__(
    self,
    arguments: Union[StrOrPath, Sequence[StrOrPath]],
    stdin: Optional[Union[bytes, File]] = None,
    stdout: Optional[File] = PIPE,
    stderr: Optional[File] = PIPE,
    buffer_size: Optional[int] = None,
) -> None:
    """Initialize a new [`Process`][process.Process] instance with the given `arguments`.

    Args:
        arguments: The command and its arguments for the process.
        stdin: A [`bytes`][bytes] object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard input.
        stdout: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard output.
        stderr: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], [`DEVNULL`][process.DEVNULL], and [`STDOUT`][process.STDOUT]) to use as the standard error.
        buffer_size: The buffer size for the stream operations. If `None`, the default buffer size will be used. If `0`, all operations will be unbuffered.

    Notes:
        If `arguments` is an instance of [`str`][str], it will be split into a list of arguments using [`shlex.split()`][shlex.split].

    Warning:
        A file-like object without a valid file descriptor, such as [`BytesIO`][io.BytesIO], cannot be used as `stdin`, `stdout`, or `stderr`.
    """
    if buffer_size is None:
        buffer_size = DEFAULT_BUFFER_SIZE

    super().__init__(arguments, stdin=stdin, stdout=stdout, stderr=stderr, buffer_size=buffer_size)

__del__ 

__del__() -> None

Clean up resources used by the process.

This method cleans up resources as follows:

  • Close the standard input stream if the process was created with stdin=PIPE and the stream is not closed.
  • If the process is running, terminate the process.
  • Wait for the process to complete.
  • Close the standard output stream if the process was created with stdout=PIPE and the stream is not closed.
  • Close the standard error stream if the process was created with stderr=PIPE and the stream is not closed.
Source code in process/process.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __del__(self) -> None:
    """Clean up resources used by the process.

    This method cleans up resources as follows:

    - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
    - If the process is running, terminate the process.
    - Wait for the process to complete.
    - Close the standard output stream if the process was created with `stdout=PIPE` and the stream is not closed.
    - Close the standard error stream if the process was created with `stderr=PIPE` and the stream is not closed.
    """
    if self._process is None:
        return

    with suppress(ProcessInvalidStreamError):
        # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
        # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
        with suppress(BrokenPipeError, ConnectionResetError):
            if not self.stdin.closed:
                self.stdin.close()

    if self.running:
        self.terminate()
        self.join()

    with suppress(ProcessInvalidStreamError):
        if not self.stdout.closed:
            self.stdout.close()

    with suppress(ProcessInvalidStreamError):
        if not self.stderr.closed:
            self.stderr.close()

run 

run() -> Self

Run the process.

Returns:

Type Description
Self

The current Process instance itself.

Raises:

Type Description
ProcessAlreadyRunError

If the process has already been run.
ProcessError

If the process fails to run.
Source code in process/process.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def run(self) -> Self:
    """Run the process.

    Returns:
        The current [`Process`][process.Process] instance itself.

    Raises:
        ProcessAlreadyRunError: If the process has already been run.
        ProcessError: If the process fails to run.
    """
    if self._process is not None:
        raise ProcessAlreadyRunError("Process has already been run")

    creationflags = 0

    # According to the documentation for [`Popen.send_signal()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/subprocess.rst?plain=1#L864),
    # > On Windows, ... CTRL_C_EVENT and CTRL_BREAK_EVENT can be sent to processes
    # > started with a *creationflags* parameter which includes ``CREATE_NEW_PROCESS_GROUP``.
    if is_windows():
        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP  # type: ignore

    stdin = self._stdin
    should_feed_stdin = isinstance(stdin, bytes)

    try:
        self._process = subprocess.Popen(
            self.arguments,
            bufsize=self._buffer_size,
            stdin=cast(File, stdin) if not should_feed_stdin else PIPE,
            stdout=self._stdout,
            stderr=self._stderr,
            creationflags=creationflags,
        )
    except Exception as exception:
        raise ProcessError(f"Failed to run process: {exception!r}")

    if should_feed_stdin:
        # Since `Popen.communicate()` requires the process to terminate, we feed the input ourselves.
        # If deadlock issues are reported, consider using a different approach, such as threading.
        self.stdin.write(cast(bytes, stdin))
        self.stdin.close()

    return self

output 

output(join: bool = True) -> bytes

Return the output from the standard output stream of the process if the process was created with stdout=PIPE.

Warning

If the standard output stream of the process is modified by others, the result of output() will be affected by those changes.

with Process("echo something") as process:
    print(process.stdout.read(4))  # b'some'
print(process.output())  # b'thing\n'

Parameters:

Name Type Description Default
join bool

Whether to wait for the process to complete before returning the output.

 True

Returns:

Type Description
bytes

The output from the standard output stream of the process if the process was created with stdout=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdout=PIPE.
Source code in process/process.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def output(self, join: bool = True) -> bytes:
    r"""Return the output from the standard output stream of the process if the process was created with `stdout=PIPE`.

    Warning:
        If the standard output stream of the process is modified by others, the result of [`output()`][process.Process.output] will be affected by those changes.

        ```python
        with Process("echo something") as process:
            print(process.stdout.read(4))  # b'some'
        print(process.output())  # b'thing\n'
        ```

    Args:
        join: Whether to wait for the process to complete before returning the output.

    Returns:
        The output from the standard output stream of the process if the process was created with `stdout=PIPE`.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
    """
    if join:
        self.join()

    if not self.stdout.closed:
        self._output += self.stdout.read()

    return self._output

error 

error(join: bool = True) -> bytes

Return the output from the standard error stream of the process if the process was created with stderr=PIPE.

Warning

If the standard error stream of the process is modified by others, the result of error() will be affected by those changes.

Parameters:

Name Type Description Default
join bool

Whether to wait for the process to complete before returning the output.

 True

Returns:

Type Description
bytes

The output from the standard error stream of the process if the process was created with stderr=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stderr=PIPE.
Source code in process/process.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def error(self, join: bool = True) -> bytes:
    """Return the output from the standard error stream of the process if the process was created with `stderr=PIPE`.

    Warning:
        If the standard error stream of the process is modified by others, the result of [`error()`][process.Process.error] will be affected by those changes.

    Args:
        join: Whether to wait for the process to complete before returning the output.

    Returns:
        The output from the standard error stream of the process if the process was created with `stderr=PIPE`.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
    """
    if join:
        self.join()

    if not self.stderr.closed:
        self._error += self.stderr.read()

    return self._error

join 

join(timeout: float | None = None) -> None

Wait for the process to complete.

Parameters:

Name Type Description Default
timeout float | None

Maximum time to wait for the process to complete, in seconds. If None, wait indefinitely.

 None

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessTimeoutError

If the process does not complete within the specified timeout.
Source code in process/process.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def join(self, timeout: Optional[float] = None) -> None:
    """Wait for the process to complete.

    Args:
        timeout: Maximum time to wait for the process to complete, in seconds. If `None`, wait indefinitely.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessTimeoutError: If the process does not complete within the specified timeout.
    """
    try:
        self.process.wait(timeout)
    except subprocess.TimeoutExpired:
        raise ProcessTimeoutError(f"Process did not complete within {timeout} seconds")

close 

close() -> None

Close the process and release its resources.

This method cleans up the process execution as follows:

  • Close the standard input stream if the process was created with stdin=PIPE and the stream is not closed.
  • If the process is running, terminate it.
  • Wait for the process to complete.
  • Read the remaining output from the standard output stream and then close the stream if the process was created with stdout=PIPE and the stream is not closed.
  • Read the remaining error from the standard error stream and then close the stream if the process was created with stderr=PIPE and the stream is not closed.
Source code in process/process.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def close(self) -> None:
    """Close the process and release its resources.

    This method cleans up the process execution as follows:

    - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
    - If the process is running, terminate it.
    - Wait for the process to complete.
    - Read the remaining output from the standard output stream and then close the stream if the process was created with `stdout=PIPE` and the stream is not closed.
    - Read the remaining error from the standard error stream and then close the stream if the process was created with `stderr=PIPE` and the stream is not closed.
    """
    with suppress(ProcessInvalidStreamError):
        # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
        # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
        with suppress(BrokenPipeError, ConnectionResetError):
            if not self.stdin.closed:
                self.stdin.close()

    self.terminate()
    self.join()

    with suppress(ProcessInvalidStreamError):
        if not self.stdout.closed:
            self._output += self.stdout.read()
            self.stdout.close()

    with suppress(ProcessInvalidStreamError):
        if not self.stderr.closed:
            self._error += self.stderr.read()
            self.stderr.close()

__enter__ 

__enter__() -> Self

Enter the runtime context for this Process instance.

This method runs the process.

Returns:

Type Description
Self

The current Process instance itself.
Source code in process/process.py 
283
284
285
286
287
288
289
290
291
def __enter__(self) -> Self:
    """Enter the runtime context for this [`Process`][process.Process] instance.

    This method runs the process.

    Returns:
        The current [`Process`][process.Process] instance itself.
    """
    return self.run()

__exit__ 

__exit__(
    exception_type: type[BaseException] | None,
    exception: BaseException | None,
    traceback: TracebackType | None,
) -> None

Exit the runtime context for this Process instance.

This method cleans up the process execution as follows:

  • Close the standard input stream if the process was created with stdin=PIPE and the stream is not closed.
  • Wait for the process to complete.
  • Read the remaining output from the standard output stream and then close the stream if the process was created with stdout=PIPE and the stream is not closed.
  • Read the remaining error from the standard error stream and then close the stream if the process was created with stderr=PIPE and the stream is not closed.
Source code in process/process.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def __exit__(
    self,
    exception_type: Optional[type[BaseException]],
    exception: Optional[BaseException],
    traceback: Optional[TracebackType],
) -> None:
    """Exit the runtime context for this [`Process`][process.Process] instance.

    This method cleans up the process execution as follows:

    - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
    - Wait for the process to complete.
    - Read the remaining output from the standard output stream and then close the stream if the process was created with `stdout=PIPE` and the stream is not closed.
    - Read the remaining error from the standard error stream and then close the stream if the process was created with `stderr=PIPE` and the stream is not closed.
    """
    with suppress(ProcessInvalidStreamError):
        # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
        # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
        with suppress(BrokenPipeError, ConnectionResetError):
            if not self.stdin.closed:
                self.stdin.close()

    if exception_type is None:
        self.join()

    with suppress(ProcessInvalidStreamError):
        if not self.stdout.closed:
            if exception_type is None:
                self._output += self.stdout.read()

            self.stdout.close()

    with suppress(ProcessInvalidStreamError):
        if not self.stderr.closed:
            if exception_type is None:
                self._error += self.stderr.read()

            self.stderr.close()

signal 

signal(signal: int) -> None

Send a signal to the process.

Parameters:

Name Type Description Default
signal int

The signal number to send.

 required

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/process.py 
194
195
196
197
198
199
200
201
202
Args:
    join: Whether to wait for the process to complete before returning the output.

Returns:
    The output from the standard output stream of the process if the process was created with `stdout=PIPE`.

Raises:
    ProcessNotRunError: If the process has not been run.
    ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.

terminate 

terminate() -> None

Gracefully terminate the process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/process.py 
205
206
207
208
209
210
211
if join:
    self.join()

if not self.stdout.closed:
    self._output += self.stdout.read()

return self._output

kill 

kill() -> None

Forcefully kill the process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/process.py 
213
214
215
216
217
218
219
def error(self, join: bool = True) -> bytes:
    """Return the output from the standard error stream of the process if the process was created with `stderr=PIPE`.

    Warning:
        If the standard error stream of the process is modified by others, the result of [`error()`][process.Process.error] will be affected by those changes.

    Args:

__repr__ 

__repr__() -> str

Return a str representation of the current process instance.

Returns:

Type Description
str

A str representation of the current process instance.
Source code in process/process.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
    self,
    exception_type: Optional[type[BaseException]],
    exception: Optional[BaseException],
    traceback: Optional[TracebackType],
) -> None:
    """Exit the runtime context for this [`Process`][process.Process] instance.

    This method cleans up the process execution as follows:

    - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
    - Wait for the process to complete.
    - Read the remaining output from the standard output stream and then close the stream if the process was created with `stdout=PIPE` and the stream is not closed.
    - Read the remaining error from the standard error stream and then close the stream if the process was created with `stderr=PIPE` and the stream is not closed.
    """
    with suppress(ProcessInvalidStreamError):
        # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
        # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
        with suppress(BrokenPipeError, ConnectionResetError):
            if not self.stdin.closed:
                self.stdin.close()

types

StreamReader module-attribute 

StreamReader: TypeAlias = FileIO | BufferedReader

StreamWriter module-attribute 

StreamWriter: TypeAlias = FileIO | BufferedWriter

process.asyncio

Process

A class for spawning, managing, and interacting with a process asynchronously.

This class provides a convenient interface for running a process, interacting with its standard input, output, and error streams, and managing its execution.

Source code in process/asyncio/process.py 
 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
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
class Process(AbstractProcess[StreamWriter, StreamReader]):
    """A class for spawning, managing, and interacting with a process asynchronously.

    This class provides a convenient interface for running a process, interacting with its standard input, output, and error streams, and managing its execution.
    """

    def __init__(
        self,
        arguments: Union[StrOrPath, Sequence[StrOrPath]],
        stdin: Optional[Union[bytes, File]] = None,
        stdout: Optional[File] = PIPE,
        stderr: Optional[File] = PIPE,
        buffer_size: Optional[int] = None,
    ) -> None:
        """Initialize a new [`Process`][process.asyncio.Process] instance with the given `arguments`.

        Args:
            arguments: The command and its arguments for the process.
            stdin: A [`bytes`][bytes] object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard input.
            stdout: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard output.
            stderr: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], [`DEVNULL`][process.DEVNULL], and [`STDOUT`][process.STDOUT]) to use as the standard error.
            buffer_size: The buffer size for the stream operations. If `None`, the default buffer size will be used. If `0`, all operations will be unbuffered.

        Notes:
            If `arguments` is an instance of [`str`][str], it will be split into a list of arguments using [`shlex.split()`][shlex.split].

        Warning:
            A file-like object without a valid file descriptor, such as [`BytesIO`][io.BytesIO], cannot be used as `stdin`, `stdout`, or `stderr`.
        """
        if buffer_size is None:
            buffer_size = DEFAULT_BUFFER_SIZE

        super().__init__(arguments, stdin=stdin, stdout=stdout, stderr=stderr, buffer_size=buffer_size)

    def __del__(self) -> None:
        """Clean up resources used by the process.

        This method cleans up resources as follows:

        - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
        - If the process is running, terminate the process.
        """
        if self._process is None:
            return

        with suppress(ProcessInvalidStreamError):
            # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
            # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
            with suppress(BrokenPipeError, ConnectionResetError):
                if not self.stdin.is_closing():
                    self.stdin.close()

        if self.running:
            self.terminate()

    @property
    def process(self) -> asyncio.subprocess.Process:
        """Return the underlying process object.

        Returns:
            An instance of [`asyncio.subprocess.Process`][asyncio.subprocess.Process].

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        if self._process is None:
            raise ProcessNotRunError("Process has not been run")

        return cast(asyncio.subprocess.Process, self._process)

    @property
    def running(self) -> bool:
        """Check if the process is currently running.

        Returns:
            `True` if the process is running, `False` otherwise.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        return self.process.returncode is None

    async def run(self) -> Self:
        """Run the process.

        Returns:
            The current [`Process`][process.asyncio.Process] instance itself.

        Raises:
            ProcessAlreadyRunError: If the process has already been run.
            ProcessError: If the process fails to run.
        """
        if self._process is not None:
            raise ProcessAlreadyRunError("Process has already been run")

        creationflags = 0

        # According to the documentation for [`Popen.send_signal()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/subprocess.rst?plain=1#L864),
        # > On Windows, ... CTRL_C_EVENT and CTRL_BREAK_EVENT can be sent to processes
        # > started with a *creationflags* parameter which includes ``CREATE_NEW_PROCESS_GROUP``.
        if is_windows():
            creationflags = subprocess.CREATE_NEW_PROCESS_GROUP  # type: ignore

        stdin = self._stdin
        should_feed_stdin = isinstance(stdin, bytes)

        try:
            self._process = await asyncio.subprocess.create_subprocess_exec(
                *self.arguments,
                stdin=cast(File, stdin) if not should_feed_stdin else PIPE,
                stdout=self._stdout,
                stderr=self._stderr,
                limit=self._buffer_size,
                creationflags=creationflags,
            )
        except Exception as exception:
            raise ProcessError(f"Failed to run process: {exception!r}")

        if should_feed_stdin:
            # Since `Popen.communicate()` requires the process to terminate, we feed the input ourselves.
            # If deadlock issues are reported, consider using a different approach, such as threading.
            self.stdin.write(cast(bytes, stdin))
            await self.stdin.drain()

            self.stdin.close()
            # Contrary to the documentation for [`StreamWriter.wait_closed()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/asyncio-stream.rst?plain=1#L383),
            # using `wait_closed()` in Python 3.7 raises `AttributeError: 'SubprocessStreamProtocol' object has no attribute '_closed'`.
            if sys.version_info >= (3, 8):
                await self.stdin.wait_closed()

        return self

    async def output(self, join: bool = True) -> bytes:
        r"""Return the output from the standard output stream of the process if the process was created with `stdout=PIPE`.

        Warning:
            If the standard output stream of the process is modified by others, the result of [`output()`][process.asyncio.Process.output] will be affected by those changes.

            ```python
            with Process("echo something") as process:
                print(process.stdout.read(4))  # b'some'
            print(process.output())  # b'thing\n'
            ```

        Args:
            join: Whether to wait for the process to complete before returning the output.

        Returns:
            The output from the standard output stream of the process if the process was created with `stdout=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
        """
        if join:
            await self.join()

        self._output += await self.stdout.read()

        return self._output

    async def error(self, join: bool = True) -> bytes:
        """Return the output from the standard error stream of the process if the process was created with `stderr=PIPE`.

        Warning:
            If the standard error stream of the process is modified by others, the result of [`error()`][process.asyncio.Process.error] will be affected by those changes.

        Args:
            join: Whether to wait for the process to complete before returning the output.

        Returns:
            The output from the standard error stream of the process if the process was created with `stderr=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
        """
        if join:
            await self.join()

        self._error += await self.stderr.read()

        return self._error

    async def join(self, timeout: Optional[float] = None) -> None:
        """Wait for the process to complete.

        Args:
            timeout: Maximum time to wait for the process to complete, in seconds. If `None`, wait indefinitely.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessTimeoutError: If the process does not complete within the specified timeout.
        """
        try:
            await asyncio.wait_for(self.process.wait(), timeout=timeout)
        except TimeoutError:
            raise ProcessTimeoutError(f"Process did not complete within {timeout} seconds")

    async def close(self) -> None:
        """Close the process and release its resources.

        This method cleans up the process execution as follows:

        - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
        - If the process is running, terminate it.
        - Wait for the process to complete.
        - Read the remaining output from the standard output stream if the process was created with `stdout=PIPE`.
        - Read the remaining error from the standard error stream if the process was created with `stderr=PIPE`.
        """
        with suppress(ProcessInvalidStreamError):
            # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
            # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
            with suppress(BrokenPipeError, ConnectionResetError):
                if not self.stdin.is_closing():
                    self.stdin.close()
                    # Contrary to the documentation for [`StreamWriter.wait_closed()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/asyncio-stream.rst?plain=1#L383),
                    # using `wait_closed()` in Python 3.7 raises `AttributeError: 'SubprocessStreamProtocol' object has no attribute '_closed'`.
                    if sys.version_info >= (3, 8):
                        await self.stdin.wait_closed()

        self.terminate()
        await self.join()

        with suppress(ProcessInvalidStreamError):
            self._output += await self.stdout.read()

        with suppress(ProcessInvalidStreamError):
            self._error += await self.stderr.read()

    async def __aenter__(self) -> Self:
        """Enter the runtime context for this [`Process`][process.asyncio.Process] instance.

        This method runs the process.

        Returns:
            The current [`Process`][process.asyncio.Process] instance itself.
        """
        return await self.run()

    async def __aexit__(
        self,
        exception_type: Optional[type[BaseException]],
        exception: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> None:
        """Exit the runtime context for this [`Process`][process.asyncio.Process] instance.

        This method cleans up the process execution as follows:

        - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
        - Wait for the process to complete.
        - Read the remaining output from the standard output stream if the process was created with `stdout=PIPE`.
        - Read the remaining error from the standard error stream if the process was created with `stderr=PIPE`.
        """
        with suppress(ProcessInvalidStreamError):
            # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
            # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
            with suppress(BrokenPipeError, ConnectionResetError):
                if not self.stdin.is_closing():
                    self.stdin.close()
                    # Contrary to the documentation for [`StreamWriter.wait_closed()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/asyncio-stream.rst?plain=1#L383),
                    # using `wait_closed()` in Python 3.7 raises `AttributeError: 'SubprocessStreamProtocol' object has no attribute '_closed'`.
                    if sys.version_info >= (3, 8):
                        await self.stdin.wait_closed()

        if exception_type is None:
            await self.join()

        with suppress(ProcessInvalidStreamError):
            if exception_type is None:
                self._output += await self.stdout.read()

        with suppress(ProcessInvalidStreamError):
            if exception_type is None:
                self._error += await self.stderr.read()

    @property
    def stdin(self) -> StreamWriter:
        """Return the standard input stream of the process if the process was created with `stdin=PIPE`.

        Returns:
            The standard input stream of the process if the process was created with `stdin=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdin=PIPE`.
        """
        stdin = self.process.stdin

        if stdin is None:
            raise ProcessInvalidStreamError("Process was not created with `stdin=PIPE`")

        return stdin

    @property
    def stdout(self) -> StreamReader:
        """Return the standard output stream of the process if the process was created with `stdout=PIPE`.

        Returns:
            The standard output stream of the process if the process was created with `stdout=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
        """
        stdout = self.process.stdout

        if stdout is None:
            raise ProcessInvalidStreamError("Process was not created with `stdout=PIPE`")

        return stdout

    @property
    def stderr(self) -> StreamReader:
        """Return the standard error stream of the process if the process was created with `stderr=PIPE`.

        Returns:
            The standard error stream of the process if the process was created with `stderr=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
        """
        stderr = self.process.stderr

        if stderr is None:
            raise ProcessInvalidStreamError("Process was not created with `stderr=PIPE`")

        return stderr

process property 

process: Process

Return the underlying process object.

Returns:

Type Description
Process

An instance of asyncio.subprocess.Process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

running property 

running: bool

Check if the process is currently running.

Returns:

Type Description
bool

True if the process is running, False otherwise.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

stdin property 

stdin: StreamWriter

Return the standard input stream of the process if the process was created with stdin=PIPE.

Returns:

Type Description
StreamWriter

The standard input stream of the process if the process was created with stdin=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdin=PIPE.

stdout property 

stdout: StreamReader

Return the standard output stream of the process if the process was created with stdout=PIPE.

Returns:

Type Description
StreamReader

The standard output stream of the process if the process was created with stdout=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdout=PIPE.

stderr property 

stderr: StreamReader

Return the standard error stream of the process if the process was created with stderr=PIPE.

Returns:

Type Description
StreamReader

The standard error stream of the process if the process was created with stderr=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stderr=PIPE.

arguments property 

arguments: list[str]

Return the command-line arguments provided to create this Process instance.

Returns:

Type Description
list[str]

The command-line arguments provided to create this Process instance.

id property 

id: int

Return the process identifier.

Returns:

Type Description
int

The process identifier.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

exit_code property 

exit_code: int | None

Return the exit code of the process.

Returns:

Type Description
int | None

The exit code of the process. If the process is still running, it returns None.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

__init__ 

__init__(
    arguments: StrOrPath | Sequence[StrOrPath],
    stdin: bytes | File | None = None,
    stdout: File | None = PIPE,
    stderr: File | None = PIPE,
    buffer_size: int | None = None,
) -> None

Initialize a new Process instance with the given arguments.

Parameters:

Name Type Description Default
arguments StrOrPath | Sequence[StrOrPath]

The command and its arguments for the process.

 required
stdin bytes | File | None

A bytes object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, and DEVNULL) to use as the standard input.

 None
stdout File | None

An existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, and DEVNULL) to use as the standard output.

 PIPE
stderr File | None

An existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, DEVNULL, and STDOUT) to use as the standard error.

 PIPE
buffer_size int | None

The buffer size for the stream operations. If None, the default buffer size will be used. If 0, all operations will be unbuffered.

 None
Notes

If arguments is an instance of str, it will be split into a list of arguments using shlex.split().

Warning

A file-like object without a valid file descriptor, such as BytesIO, cannot be used as stdin, stdout, or stderr.

Source code in process/asyncio/process.py 
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
def __init__(
    self,
    arguments: Union[StrOrPath, Sequence[StrOrPath]],
    stdin: Optional[Union[bytes, File]] = None,
    stdout: Optional[File] = PIPE,
    stderr: Optional[File] = PIPE,
    buffer_size: Optional[int] = None,
) -> None:
    """Initialize a new [`Process`][process.asyncio.Process] instance with the given `arguments`.

    Args:
        arguments: The command and its arguments for the process.
        stdin: A [`bytes`][bytes] object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard input.
        stdout: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard output.
        stderr: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], [`DEVNULL`][process.DEVNULL], and [`STDOUT`][process.STDOUT]) to use as the standard error.
        buffer_size: The buffer size for the stream operations. If `None`, the default buffer size will be used. If `0`, all operations will be unbuffered.

    Notes:
        If `arguments` is an instance of [`str`][str], it will be split into a list of arguments using [`shlex.split()`][shlex.split].

    Warning:
        A file-like object without a valid file descriptor, such as [`BytesIO`][io.BytesIO], cannot be used as `stdin`, `stdout`, or `stderr`.
    """
    if buffer_size is None:
        buffer_size = DEFAULT_BUFFER_SIZE

    super().__init__(arguments, stdin=stdin, stdout=stdout, stderr=stderr, buffer_size=buffer_size)

__del__ 

__del__() -> None

Clean up resources used by the process.

This method cleans up resources as follows:

  • Close the standard input stream if the process was created with stdin=PIPE and the stream is not closed.
  • If the process is running, terminate the process.
Source code in process/asyncio/process.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __del__(self) -> None:
    """Clean up resources used by the process.

    This method cleans up resources as follows:

    - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
    - If the process is running, terminate the process.
    """
    if self._process is None:
        return

    with suppress(ProcessInvalidStreamError):
        # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
        # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
        with suppress(BrokenPipeError, ConnectionResetError):
            if not self.stdin.is_closing():
                self.stdin.close()

    if self.running:
        self.terminate()

run async 

run() -> Self

Run the process.

Returns:

Type Description
Self

The current Process instance itself.

Raises:

Type Description
ProcessAlreadyRunError

If the process has already been run.
ProcessError

If the process fails to run.
Source code in process/asyncio/process.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
async def run(self) -> Self:
    """Run the process.

    Returns:
        The current [`Process`][process.asyncio.Process] instance itself.

    Raises:
        ProcessAlreadyRunError: If the process has already been run.
        ProcessError: If the process fails to run.
    """
    if self._process is not None:
        raise ProcessAlreadyRunError("Process has already been run")

    creationflags = 0

    # According to the documentation for [`Popen.send_signal()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/subprocess.rst?plain=1#L864),
    # > On Windows, ... CTRL_C_EVENT and CTRL_BREAK_EVENT can be sent to processes
    # > started with a *creationflags* parameter which includes ``CREATE_NEW_PROCESS_GROUP``.
    if is_windows():
        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP  # type: ignore

    stdin = self._stdin
    should_feed_stdin = isinstance(stdin, bytes)

    try:
        self._process = await asyncio.subprocess.create_subprocess_exec(
            *self.arguments,
            stdin=cast(File, stdin) if not should_feed_stdin else PIPE,
            stdout=self._stdout,
            stderr=self._stderr,
            limit=self._buffer_size,
            creationflags=creationflags,
        )
    except Exception as exception:
        raise ProcessError(f"Failed to run process: {exception!r}")

    if should_feed_stdin:
        # Since `Popen.communicate()` requires the process to terminate, we feed the input ourselves.
        # If deadlock issues are reported, consider using a different approach, such as threading.
        self.stdin.write(cast(bytes, stdin))
        await self.stdin.drain()

        self.stdin.close()
        # Contrary to the documentation for [`StreamWriter.wait_closed()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/asyncio-stream.rst?plain=1#L383),
        # using `wait_closed()` in Python 3.7 raises `AttributeError: 'SubprocessStreamProtocol' object has no attribute '_closed'`.
        if sys.version_info >= (3, 8):
            await self.stdin.wait_closed()

    return self

output async 

output(join: bool = True) -> bytes

Return the output from the standard output stream of the process if the process was created with stdout=PIPE.

Warning

If the standard output stream of the process is modified by others, the result of output() will be affected by those changes.

with Process("echo something") as process:
    print(process.stdout.read(4))  # b'some'
print(process.output())  # b'thing\n'

Parameters:

Name Type Description Default
join bool

Whether to wait for the process to complete before returning the output.

 True

Returns:

Type Description
bytes

The output from the standard output stream of the process if the process was created with stdout=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdout=PIPE.
Source code in process/asyncio/process.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
async def output(self, join: bool = True) -> bytes:
    r"""Return the output from the standard output stream of the process if the process was created with `stdout=PIPE`.

    Warning:
        If the standard output stream of the process is modified by others, the result of [`output()`][process.asyncio.Process.output] will be affected by those changes.

        ```python
        with Process("echo something") as process:
            print(process.stdout.read(4))  # b'some'
        print(process.output())  # b'thing\n'
        ```

    Args:
        join: Whether to wait for the process to complete before returning the output.

    Returns:
        The output from the standard output stream of the process if the process was created with `stdout=PIPE`.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
    """
    if join:
        await self.join()

    self._output += await self.stdout.read()

    return self._output

error async 

error(join: bool = True) -> bytes

Return the output from the standard error stream of the process if the process was created with stderr=PIPE.

Warning

If the standard error stream of the process is modified by others, the result of error() will be affected by those changes.

Parameters:

Name Type Description Default
join bool

Whether to wait for the process to complete before returning the output.

 True

Returns:

Type Description
bytes

The output from the standard error stream of the process if the process was created with stderr=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stderr=PIPE.
Source code in process/asyncio/process.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
async def error(self, join: bool = True) -> bytes:
    """Return the output from the standard error stream of the process if the process was created with `stderr=PIPE`.

    Warning:
        If the standard error stream of the process is modified by others, the result of [`error()`][process.asyncio.Process.error] will be affected by those changes.

    Args:
        join: Whether to wait for the process to complete before returning the output.

    Returns:
        The output from the standard error stream of the process if the process was created with `stderr=PIPE`.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
    """
    if join:
        await self.join()

    self._error += await self.stderr.read()

    return self._error

join async 

join(timeout: float | None = None) -> None

Wait for the process to complete.

Parameters:

Name Type Description Default
timeout float | None

Maximum time to wait for the process to complete, in seconds. If None, wait indefinitely.

 None

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessTimeoutError

If the process does not complete within the specified timeout.
Source code in process/asyncio/process.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
async def join(self, timeout: Optional[float] = None) -> None:
    """Wait for the process to complete.

    Args:
        timeout: Maximum time to wait for the process to complete, in seconds. If `None`, wait indefinitely.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessTimeoutError: If the process does not complete within the specified timeout.
    """
    try:
        await asyncio.wait_for(self.process.wait(), timeout=timeout)
    except TimeoutError:
        raise ProcessTimeoutError(f"Process did not complete within {timeout} seconds")

close async 

close() -> None

Close the process and release its resources.

This method cleans up the process execution as follows:

  • Close the standard input stream if the process was created with stdin=PIPE and the stream is not closed.
  • If the process is running, terminate it.
  • Wait for the process to complete.
  • Read the remaining output from the standard output stream if the process was created with stdout=PIPE.
  • Read the remaining error from the standard error stream if the process was created with stderr=PIPE.
Source code in process/asyncio/process.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
async def close(self) -> None:
    """Close the process and release its resources.

    This method cleans up the process execution as follows:

    - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
    - If the process is running, terminate it.
    - Wait for the process to complete.
    - Read the remaining output from the standard output stream if the process was created with `stdout=PIPE`.
    - Read the remaining error from the standard error stream if the process was created with `stderr=PIPE`.
    """
    with suppress(ProcessInvalidStreamError):
        # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
        # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
        with suppress(BrokenPipeError, ConnectionResetError):
            if not self.stdin.is_closing():
                self.stdin.close()
                # Contrary to the documentation for [`StreamWriter.wait_closed()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/asyncio-stream.rst?plain=1#L383),
                # using `wait_closed()` in Python 3.7 raises `AttributeError: 'SubprocessStreamProtocol' object has no attribute '_closed'`.
                if sys.version_info >= (3, 8):
                    await self.stdin.wait_closed()

    self.terminate()
    await self.join()

    with suppress(ProcessInvalidStreamError):
        self._output += await self.stdout.read()

    with suppress(ProcessInvalidStreamError):
        self._error += await self.stderr.read()

__aenter__ async 

__aenter__() -> Self

Enter the runtime context for this Process instance.

This method runs the process.

Returns:

Type Description
Self

The current Process instance itself.
Source code in process/asyncio/process.py 
260
261
262
263
264
265
266
267
268
async def __aenter__(self) -> Self:
    """Enter the runtime context for this [`Process`][process.asyncio.Process] instance.

    This method runs the process.

    Returns:
        The current [`Process`][process.asyncio.Process] instance itself.
    """
    return await self.run()

__aexit__ async 

__aexit__(
    exception_type: type[BaseException] | None,
    exception: BaseException | None,
    traceback: TracebackType | None,
) -> None

Exit the runtime context for this Process instance.

This method cleans up the process execution as follows:

  • Close the standard input stream if the process was created with stdin=PIPE and the stream is not closed.
  • Wait for the process to complete.
  • Read the remaining output from the standard output stream if the process was created with stdout=PIPE.
  • Read the remaining error from the standard error stream if the process was created with stderr=PIPE.
Source code in process/asyncio/process.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
async def __aexit__(
    self,
    exception_type: Optional[type[BaseException]],
    exception: Optional[BaseException],
    traceback: Optional[TracebackType],
) -> None:
    """Exit the runtime context for this [`Process`][process.asyncio.Process] instance.

    This method cleans up the process execution as follows:

    - Close the standard input stream if the process was created with `stdin=PIPE` and the stream is not closed.
    - Wait for the process to complete.
    - Read the remaining output from the standard output stream if the process was created with `stdout=PIPE`.
    - Read the remaining error from the standard error stream if the process was created with `stderr=PIPE`.
    """
    with suppress(ProcessInvalidStreamError):
        # Since we are cleaning up, suppress errors from closing the standard input stream by following the approach in
        # https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/subprocess.py#L1097
        with suppress(BrokenPipeError, ConnectionResetError):
            if not self.stdin.is_closing():
                self.stdin.close()
                # Contrary to the documentation for [`StreamWriter.wait_closed()`](https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Doc/library/asyncio-stream.rst?plain=1#L383),
                # using `wait_closed()` in Python 3.7 raises `AttributeError: 'SubprocessStreamProtocol' object has no attribute '_closed'`.
                if sys.version_info >= (3, 8):
                    await self.stdin.wait_closed()

    if exception_type is None:
        await self.join()

    with suppress(ProcessInvalidStreamError):
        if exception_type is None:
            self._output += await self.stdout.read()

    with suppress(ProcessInvalidStreamError):
        if exception_type is None:
            self._error += await self.stderr.read()

signal 

signal(signal: int) -> None

Send a signal to the process.

Parameters:

Name Type Description Default
signal int

The signal number to send.

 required

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/asyncio/process.py 
194
195
196
197
198
199
200
201
202
203
Warning:
    If the standard error stream of the process is modified by others, the result of [`error()`][process.asyncio.Process.error] will be affected by those changes.

Args:
    join: Whether to wait for the process to complete before returning the output.

Returns:
    The output from the standard error stream of the process if the process was created with `stderr=PIPE`.

Raises:

terminate 

terminate() -> None

Gracefully terminate the process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/asyncio/process.py 
205
206
207
208
209
210
    ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
"""
if join:
    await self.join()

self._error += await self.stderr.read()

kill 

kill() -> None

Forcefully kill the process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/asyncio/process.py 
213
214
215
216
217
async def join(self, timeout: Optional[float] = None) -> None:
    """Wait for the process to complete.

    Args:
        timeout: Maximum time to wait for the process to complete, in seconds. If `None`, wait indefinitely.

__repr__ 

__repr__() -> str

Return a str representation of the current process instance.

Returns:

Type Description
str

A str representation of the current process instance.
Source code in process/asyncio/process.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
                    await self.stdin.wait_closed()

    if exception_type is None:
        await self.join()

    with suppress(ProcessInvalidStreamError):
        if exception_type is None:
            self._output += await self.stdout.read()

    with suppress(ProcessInvalidStreamError):
        if exception_type is None:
            self._error += await self.stderr.read()

@property
def stdin(self) -> StreamWriter:
    """Return the standard input stream of the process if the process was created with `stdin=PIPE`.

    Returns:
        The standard input stream of the process if the process was created with `stdin=PIPE`.

    Raises:

types

StreamReader module-attribute 

StreamReader: TypeAlias = StreamReader

StreamWriter module-attribute 

StreamWriter: TypeAlias = StreamWriter

core

AbstractProcess

Abstract base class for managing processes.

Source code in process/core/process.py 
 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
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
class AbstractProcess(ABC, ProcessProtocol[W, R], Generic[W, R]):
    """Abstract base class for managing processes."""

    def __init__(
        self,
        arguments: Union[StrOrPath, Sequence[StrOrPath]],
        stdin: Optional[Union[bytes, File]] = None,
        stdout: Optional[File] = PIPE,
        stderr: Optional[File] = PIPE,
        buffer_size: Optional[int] = None,
    ) -> None:
        """Initialize a new [`Process`][process.core.process.AbstractProcess] instance with the given `arguments`.

        Args:
            arguments: The command and its arguments for the process.
            stdin: A [`bytes`][bytes] object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard input.
            stdout: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard output.
            stderr: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], [`DEVNULL`][process.DEVNULL], and [`STDOUT`][process.STDOUT]) to use as the standard error.
            buffer_size: The buffer size for the stream operations. If `None`, the default buffer size will be used. If `0`, all operations will be unbuffered.

        Notes:
            If `arguments` is an instance of [`str`][str], it will be split into a list of arguments using [`shlex.split()`][shlex.split].

        Warning:
            A file-like object without a valid file descriptor, such as [`BytesIO`][io.BytesIO], cannot be used as `stdin`, `stdout`, or `stderr`.
        """
        if isinstance(arguments, str):
            arguments = shlex.split(arguments)
        elif isinstance(arguments, PathLike):
            arguments = [arguments]

        if buffer_size is None:
            raise ValueError("`buffer_size` should be set to the default buffer size.")

        self._arguments: list[str] = [fspath(argument) for argument in arguments]

        self._stdin: Optional[Union[bytes, File]] = stdin
        self._stdout: Optional[File] = stdout
        self._stderr: Optional[File] = stderr

        self._buffer_size: int = buffer_size

        self._output: bytes = b""
        self._error: bytes = b""

        if sys.version_info >= (3, 9):
            self._process: Optional[Union[subprocess.Popen[bytes], asyncio.subprocess.Process]] = None
        else:
            self._process: Optional[Union[subprocess.Popen, asyncio.subprocess.Process]] = None

    @abstractmethod
    def __del__(self) -> None:
        """Clean up resources used by the process."""
        ...

    @property
    def arguments(self) -> list[str]:
        """Return the command-line arguments provided to create this [`Process`][process.core.process.AbstractProcess] instance.

        Returns:
            The command-line arguments provided to create this [`Process`][process.core.process.AbstractProcess] instance.
        """
        return list(self._arguments)

    if sys.version_info >= (3, 9):

        @property
        @abstractmethod
        def process(self) -> Union[subprocess.Popen[bytes], asyncio.subprocess.Process]:
            """Return the underlying process object.

            Returns:
                An instance of either [`subprocess.Popen`][subprocess.Popen] or [`asyncio.subprocess.Process`][asyncio.subprocess.Process].

            Raises:
                ProcessNotRunError: If the process has not been run.
            """
            ...
    else:

        @property
        @abstractmethod
        def process(self) -> Union[subprocess.Popen, asyncio.subprocess.Process]:
            """Return the underlying process object.

            Returns:
                An instance of either [`subprocess.Popen`][subprocess.Popen] or [`asyncio.subprocess.Process`][asyncio.subprocess.Process].

            Raises:
                ProcessNotRunError: If the process has not been run.
            """
            ...

    @property
    def id(self) -> int:
        """Return the process identifier.

        Returns:
            The process identifier.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        return self.process.pid

    @property
    @abstractmethod
    def running(self) -> bool:
        """Check if the process is currently running.

        Returns:
            `True` if the process is running, `False` otherwise.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        ...

    @property
    def exit_code(self) -> Optional[int]:
        """Return the exit code of the process.

        Returns:
            The exit code of the process. If the process is still running, it returns `None`.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        return self.process.returncode

    @abstractmethod
    def run(self) -> Returns[Self]:
        """Run the process.

        Returns:
            The current [`Process`][process.core.process.AbstractProcess] instance itself.

        Raises:
            ProcessAlreadyRunError: If the process has already been run.
            ProcessError: If the process fails to run.
        """
        ...

    @abstractmethod
    def output(self, join: bool = True) -> Returns[bytes]:
        """Return the output from the standard output stream of the process if the process was created with `stdout=PIPE`.

        Args:
            join: Whether to wait for the process to complete before returning the output.

        Returns:
            The output from the standard output stream of the process if the process was created with `stdout=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
        """
        ...

    @abstractmethod
    def error(self, join: bool = True) -> Returns[bytes]:
        """Return the output from the standard error stream of the process if the process was created with `stderr=PIPE`.

        Args:
            join: Whether to wait for the process to complete before returning the output.

        Returns:
            The output from the standard error stream of the process if the process was created with `stderr=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
        """
        ...

    @abstractmethod
    def join(self, timeout: Optional[float] = None) -> Returns[None]:
        """Wait for the process to complete.

        Args:
            timeout: Maximum time to wait for the process to complete, in seconds. If `None`, wait indefinitely.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessTimeoutError: If the process does not complete within the specified timeout.
        """
        ...

    def signal(self, signal: int) -> None:
        """Send a signal to the process.

        Args:
            signal: The signal number to send.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        self.process.send_signal(signal)

    def terminate(self) -> None:
        """Gracefully terminate the process.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        self.process.terminate()

    def kill(self) -> None:
        """Forcefully kill the process.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        self.process.kill()

    @abstractmethod
    def close(self) -> Returns[None]:
        """Close the process and release its resources.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """

    @property
    @abstractmethod
    def stdin(self) -> W:
        """Return the standard input stream of the process if the process was created with `stdin=PIPE`.

        Returns:
            The standard input stream of the process if the process was created with `stdin=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdin=PIPE`.
        """
        ...

    @property
    @abstractmethod
    def stdout(self) -> R:
        """Return the standard output stream of the process if the process was created with `stdout=PIPE`.

        Returns:
            The standard output stream of the process if the process was created with `stdout=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
        """
        ...

    @property
    @abstractmethod
    def stderr(self) -> R:
        """Return the standard error stream of the process if the process was created with `stderr=PIPE`.

        Returns:
            The standard error stream of the process if the process was created with `stderr=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
        """
        ...

    def __repr__(self) -> str:
        """Return a [`str`][str] representation of the current process instance.

        Returns:
            A [`str`][str] representation of the current process instance.
        """
        items = [f"arguments={self.arguments!r}"]

        with suppress(ProcessNotRunError):
            items.extend([f"id={self.id!r}", f"running={self.running!r}", f"exit_code={self.exit_code!r}"])

            with suppress(ProcessInvalidStreamError):
                items.append(f"stdin={self.stdin!r}")

            with suppress(ProcessInvalidStreamError):
                items.append(f"stdout={self.stdout!r}")

            with suppress(ProcessInvalidStreamError):
                items.append(f"stderr={self.stderr!r}")

        return f"{type(self).__name__}({', '.join(items)})"

arguments property 

arguments: list[str]

Return the command-line arguments provided to create this Process instance.

Returns:

Type Description
list[str]

The command-line arguments provided to create this Process instance.

process abstractmethod property 

process: Popen | Process

Return the underlying process object.

Returns:

Type Description
Popen | Process

An instance of either subprocess.Popen or asyncio.subprocess.Process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

id property 

id: int

Return the process identifier.

Returns:

Type Description
int

The process identifier.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

running abstractmethod property 

running: bool

Check if the process is currently running.

Returns:

Type Description
bool

True if the process is running, False otherwise.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

exit_code property 

exit_code: int | None

Return the exit code of the process.

Returns:

Type Description
int | None

The exit code of the process. If the process is still running, it returns None.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

stdin abstractmethod property 

stdin: W

Return the standard input stream of the process if the process was created with stdin=PIPE.

Returns:

Type Description
W

The standard input stream of the process if the process was created with stdin=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdin=PIPE.

stdout abstractmethod property 

stdout: R

Return the standard output stream of the process if the process was created with stdout=PIPE.

Returns:

Type Description
R

The standard output stream of the process if the process was created with stdout=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdout=PIPE.

stderr abstractmethod property 

stderr: R

Return the standard error stream of the process if the process was created with stderr=PIPE.

Returns:

Type Description
R

The standard error stream of the process if the process was created with stderr=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stderr=PIPE.

__init__ 

__init__(
    arguments: StrOrPath | Sequence[StrOrPath],
    stdin: bytes | File | None = None,
    stdout: File | None = PIPE,
    stderr: File | None = PIPE,
    buffer_size: int | None = None,
) -> None

Initialize a new Process instance with the given arguments.

Parameters:

Name Type Description Default
arguments StrOrPath | Sequence[StrOrPath]

The command and its arguments for the process.

 required
stdin bytes | File | None

A bytes object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, and DEVNULL) to use as the standard input.

 None
stdout File | None

An existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, and DEVNULL) to use as the standard output.

 PIPE
stderr File | None

An existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, DEVNULL, and STDOUT) to use as the standard error.

 PIPE
buffer_size int | None

The buffer size for the stream operations. If None, the default buffer size will be used. If 0, all operations will be unbuffered.

 None
Notes

If arguments is an instance of str, it will be split into a list of arguments using shlex.split().

Warning

A file-like object without a valid file descriptor, such as BytesIO, cannot be used as stdin, stdout, or stderr.

Source code in process/core/process.py 
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
def __init__(
    self,
    arguments: Union[StrOrPath, Sequence[StrOrPath]],
    stdin: Optional[Union[bytes, File]] = None,
    stdout: Optional[File] = PIPE,
    stderr: Optional[File] = PIPE,
    buffer_size: Optional[int] = None,
) -> None:
    """Initialize a new [`Process`][process.core.process.AbstractProcess] instance with the given `arguments`.

    Args:
        arguments: The command and its arguments for the process.
        stdin: A [`bytes`][bytes] object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard input.
        stdout: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard output.
        stderr: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], [`DEVNULL`][process.DEVNULL], and [`STDOUT`][process.STDOUT]) to use as the standard error.
        buffer_size: The buffer size for the stream operations. If `None`, the default buffer size will be used. If `0`, all operations will be unbuffered.

    Notes:
        If `arguments` is an instance of [`str`][str], it will be split into a list of arguments using [`shlex.split()`][shlex.split].

    Warning:
        A file-like object without a valid file descriptor, such as [`BytesIO`][io.BytesIO], cannot be used as `stdin`, `stdout`, or `stderr`.
    """
    if isinstance(arguments, str):
        arguments = shlex.split(arguments)
    elif isinstance(arguments, PathLike):
        arguments = [arguments]

    if buffer_size is None:
        raise ValueError("`buffer_size` should be set to the default buffer size.")

    self._arguments: list[str] = [fspath(argument) for argument in arguments]

    self._stdin: Optional[Union[bytes, File]] = stdin
    self._stdout: Optional[File] = stdout
    self._stderr: Optional[File] = stderr

    self._buffer_size: int = buffer_size

    self._output: bytes = b""
    self._error: bytes = b""

    if sys.version_info >= (3, 9):
        self._process: Optional[Union[subprocess.Popen[bytes], asyncio.subprocess.Process]] = None
    else:
        self._process: Optional[Union[subprocess.Popen, asyncio.subprocess.Process]] = None

__del__ abstractmethod 

__del__() -> None

Clean up resources used by the process.

Source code in process/core/process.py 
79
80
81
82
@abstractmethod
def __del__(self) -> None:
    """Clean up resources used by the process."""
    ...

run abstractmethod 

run() -> Returns[Self]

Run the process.

Returns:

Type Description
Returns[Self]

The current Process instance itself.

Raises:

Type Description
ProcessAlreadyRunError

If the process has already been run.
ProcessError

If the process fails to run.
Source code in process/core/process.py 
159
160
161
162
163
164
165
166
167
168
169
170
@abstractmethod
def run(self) -> Returns[Self]:
    """Run the process.

    Returns:
        The current [`Process`][process.core.process.AbstractProcess] instance itself.

    Raises:
        ProcessAlreadyRunError: If the process has already been run.
        ProcessError: If the process fails to run.
    """
    ...

output abstractmethod 

output(join: bool = True) -> Returns[bytes]

Return the output from the standard output stream of the process if the process was created with stdout=PIPE.

Parameters:

Name Type Description Default
join bool

Whether to wait for the process to complete before returning the output.

 True

Returns:

Type Description
Returns[bytes]

The output from the standard output stream of the process if the process was created with stdout=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdout=PIPE.
Source code in process/core/process.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@abstractmethod
def output(self, join: bool = True) -> Returns[bytes]:
    """Return the output from the standard output stream of the process if the process was created with `stdout=PIPE`.

    Args:
        join: Whether to wait for the process to complete before returning the output.

    Returns:
        The output from the standard output stream of the process if the process was created with `stdout=PIPE`.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
    """
    ...

error abstractmethod 

error(join: bool = True) -> Returns[bytes]

Return the output from the standard error stream of the process if the process was created with stderr=PIPE.

Parameters:

Name Type Description Default
join bool

Whether to wait for the process to complete before returning the output.

 True

Returns:

Type Description
Returns[bytes]

The output from the standard error stream of the process if the process was created with stderr=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stderr=PIPE.
Source code in process/core/process.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
@abstractmethod
def error(self, join: bool = True) -> Returns[bytes]:
    """Return the output from the standard error stream of the process if the process was created with `stderr=PIPE`.

    Args:
        join: Whether to wait for the process to complete before returning the output.

    Returns:
        The output from the standard error stream of the process if the process was created with `stderr=PIPE`.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
    """
    ...

join abstractmethod 

join(timeout: float | None = None) -> Returns[None]

Wait for the process to complete.

Parameters:

Name Type Description Default
timeout float | None

Maximum time to wait for the process to complete, in seconds. If None, wait indefinitely.

 None

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessTimeoutError

If the process does not complete within the specified timeout.
Source code in process/core/process.py 
204
205
206
207
208
209
210
211
212
213
214
215
@abstractmethod
def join(self, timeout: Optional[float] = None) -> Returns[None]:
    """Wait for the process to complete.

    Args:
        timeout: Maximum time to wait for the process to complete, in seconds. If `None`, wait indefinitely.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessTimeoutError: If the process does not complete within the specified timeout.
    """
    ...

signal 

signal(signal: int) -> None

Send a signal to the process.

Parameters:

Name Type Description Default
signal int

The signal number to send.

 required

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/core/process.py 
217
218
219
220
221
222
223
224
225
226
def signal(self, signal: int) -> None:
    """Send a signal to the process.

    Args:
        signal: The signal number to send.

    Raises:
        ProcessNotRunError: If the process has not been run.
    """
    self.process.send_signal(signal)

terminate 

terminate() -> None

Gracefully terminate the process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/core/process.py 
228
229
230
231
232
233
234
def terminate(self) -> None:
    """Gracefully terminate the process.

    Raises:
        ProcessNotRunError: If the process has not been run.
    """
    self.process.terminate()

kill 

kill() -> None

Forcefully kill the process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/core/process.py 
236
237
238
239
240
241
242
def kill(self) -> None:
    """Forcefully kill the process.

    Raises:
        ProcessNotRunError: If the process has not been run.
    """
    self.process.kill()

close abstractmethod 

close() -> Returns[None]

Close the process and release its resources.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/core/process.py 
244
245
246
247
248
249
250
@abstractmethod
def close(self) -> Returns[None]:
    """Close the process and release its resources.

    Raises:
        ProcessNotRunError: If the process has not been run.
    """

__repr__ 

__repr__() -> str

Return a str representation of the current process instance.

Returns:

Type Description
str

A str representation of the current process instance.
Source code in process/core/process.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
def __repr__(self) -> str:
    """Return a [`str`][str] representation of the current process instance.

    Returns:
        A [`str`][str] representation of the current process instance.
    """
    items = [f"arguments={self.arguments!r}"]

    with suppress(ProcessNotRunError):
        items.extend([f"id={self.id!r}", f"running={self.running!r}", f"exit_code={self.exit_code!r}"])

        with suppress(ProcessInvalidStreamError):
            items.append(f"stdin={self.stdin!r}")

        with suppress(ProcessInvalidStreamError):
            items.append(f"stdout={self.stdout!r}")

        with suppress(ProcessInvalidStreamError):
            items.append(f"stderr={self.stderr!r}")

    return f"{type(self).__name__}({', '.join(items)})"

ProcessProtocol

A protocol that defines the interface for spawning and managing a process.

Source code in process/core/protocol.py 
 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
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
class ProcessProtocol(Protocol[W_co, R_co]):
    """A protocol that defines the interface for spawning and managing a process."""

    @overload
    def __init__(
        self,
        arguments: StrOrPath,
        stdin: Optional[Union[bytes, File]] = None,
        stdout: Optional[File] = PIPE,
        stderr: Optional[File] = PIPE,
        buffer_size: Optional[int] = None,
    ) -> None: ...
    @overload
    def __init__(
        self,
        arguments: Sequence[StrOrPath],
        stdin: Optional[Union[bytes, File]] = None,
        stdout: Optional[File] = PIPE,
        stderr: Optional[File] = PIPE,
        buffer_size: Optional[int] = None,
    ) -> None: ...

    def __init__(
        self,
        arguments: Union[StrOrPath, Sequence[StrOrPath]],
        stdin: Optional[Union[bytes, File]] = None,
        stdout: Optional[File] = PIPE,
        stderr: Optional[File] = PIPE,
        buffer_size: Optional[int] = None,
    ) -> None:
        """Initialize a new `Process` instance with the given `arguments`.

        Args:
            arguments: The command and its arguments for the process.
            stdin: A [`bytes`][bytes] object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard input.
            stdout: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard output.
            stderr: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], [`DEVNULL`][process.DEVNULL], and [`STDOUT`][process.STDOUT]) to use as the standard error.
            buffer_size: The buffer size for the stream operations. If `None`, the default buffer size will be used. If `0`, all operations will be unbuffered.

        Notes:
            If `arguments` is an instance of [`str`][str], it will be split into a list of arguments using [`shlex.split()`][shlex.split].

        Warning:
            A file-like object without a valid file descriptor, such as [`BytesIO`][io.BytesIO], cannot be used as `stdin`, `stdout`, or `stderr`.
        """
        ...

    def __del__(self) -> None:
        """Clean up resources used by the process."""
        ...

    @property
    def arguments(self) -> list[str]:
        """Return the command-line arguments provided to create this `Process` instance.

        Returns:
            The command-line arguments provided to create this `Process` instance.
        """
        ...

    if sys.version_info >= (3, 9):

        @property
        def process(self) -> Union[subprocess.Popen[bytes], asyncio.subprocess.Process]:
            """Return the underlying process instance.

            Returns:
                An instance of either [`subprocess.Popen`][subprocess.Popen] or [`asyncio.subprocess.Process`][asyncio.subprocess.Process].

            Raises:
                ProcessNotRunError: If the process has not been run.
            """
            ...
    else:

        @property
        def process(self) -> Union[subprocess.Popen, asyncio.subprocess.Process]:
            """Return the underlying process instance.

            Returns:
                An instance of either [`subprocess.Popen`][subprocess.Popen] or [`asyncio.subprocess.Process`][asyncio.subprocess.Process].

            Raises:
                ProcessNotRunError: If the process has not been run.
            """
            ...

    @property
    def id(self) -> int:
        """Return the process identifier.

        Returns:
            The process identifier.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        ...

    @property
    def running(self) -> bool:
        """Check if the process is currently running.

        Returns:
            `True` if the process is running, `False` otherwise.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        ...

    @property
    def exit_code(self) -> Optional[int]:
        """Return the exit code of the process.

        Returns:
            The exit code of the process. If the process is still running, it returns `None`.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        ...

    def run(self) -> Returns[Self]:
        """Run the process.

        Returns:
            The current `Process` instance itself.

        Raises:
            ProcessAlreadyRunError: If the process has already been run.
            ProcessError: If the process fails to run.
        """
        ...

    def output(self, join: bool = True) -> Returns[bytes]:
        """Return the output from the standard output stream of the process if the process was created with `stdout=PIPE`.

        Args:
            join: Whether to wait for the process to complete before returning the output.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
        """
        ...

    def error(self, join: bool = True) -> Returns[bytes]:
        """Return the output from the standard error stream of the process if the process was created with `stderr=PIPE`.

        Args:
            join: Whether to wait for the process to complete before returning the output.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
        """
        ...

    def join(self, timeout: Optional[float] = None) -> Returns[None]:
        """Wait for the process to complete.

        Args:
            timeout: Maximum time to wait for the process to complete, in seconds. If `None`, wait indefinitely.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessTimeoutError: If the process does not complete within the specified timeout.
        """
        ...

    def signal(self, signal: int) -> None:
        """Send a signal to the process.

        Args:
            signal: The signal number to send.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        ...

    def terminate(self) -> None:
        """Gracefully terminate the process.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        ...

    def kill(self) -> None:
        """Forcefully kill the process.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """
        ...

    def close(self) -> Returns[None]:
        """Close the process and release its resources.

        Raises:
            ProcessNotRunError: If the process has not been run.
        """

    @property
    def stdin(self) -> W_co:
        """Return the standard input stream of the process.

        Returns:
            The standard input stream of the process if the process was created with `stdin=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdin=PIPE`.
        """
        ...

    @property
    def stdout(self) -> R_co:
        """Return the standard output stream of the process.

        Returns:
            The standard output stream of the process if the process was created with `stdout=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
        """
        ...

    @property
    def stderr(self) -> R_co:
        """Return the standard error stream of the process.

        Returns:
            The standard error stream of the process if the process was created with `stderr=PIPE`.

        Raises:
            ProcessNotRunError: If the process has not been run.
            ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
        """
        ...

arguments property 

arguments: list[str]

Return the command-line arguments provided to create this Process instance.

Returns:

Type Description
list[str]

The command-line arguments provided to create this Process instance.

process property 

process: Popen | Process

Return the underlying process instance.

Returns:

Type Description
Popen | Process

An instance of either subprocess.Popen or asyncio.subprocess.Process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

id property 

id: int

Return the process identifier.

Returns:

Type Description
int

The process identifier.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

running property 

running: bool

Check if the process is currently running.

Returns:

Type Description
bool

True if the process is running, False otherwise.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

exit_code property 

exit_code: int | None

Return the exit code of the process.

Returns:

Type Description
int | None

The exit code of the process. If the process is still running, it returns None.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.

stdin property 

stdin: W_co

Return the standard input stream of the process.

Returns:

Type Description
W_co

The standard input stream of the process if the process was created with stdin=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdin=PIPE.

stdout property 

stdout: R_co

Return the standard output stream of the process.

Returns:

Type Description
R_co

The standard output stream of the process if the process was created with stdout=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdout=PIPE.

stderr property 

stderr: R_co

Return the standard error stream of the process.

Returns:

Type Description
R_co

The standard error stream of the process if the process was created with stderr=PIPE.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stderr=PIPE.

__init__ 

__init__(
    arguments: StrOrPath | Sequence[StrOrPath],
    stdin: bytes | File | None = None,
    stdout: File | None = PIPE,
    stderr: File | None = PIPE,
    buffer_size: int | None = None,
) -> None

Initialize a new Process instance with the given arguments.

Parameters:

Name Type Description Default
arguments StrOrPath | Sequence[StrOrPath]

The command and its arguments for the process.

 required
stdin bytes | File | None

A bytes object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, and DEVNULL) to use as the standard input.

 None
stdout File | None

An existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, and DEVNULL) to use as the standard output.

 PIPE
stderr File | None

An existing file descriptor, an existing file object with a valid file descriptor, or special value (None, PIPE, DEVNULL, and STDOUT) to use as the standard error.

 PIPE
buffer_size int | None

The buffer size for the stream operations. If None, the default buffer size will be used. If 0, all operations will be unbuffered.

 None
Notes

If arguments is an instance of str, it will be split into a list of arguments using shlex.split().

Warning

A file-like object without a valid file descriptor, such as BytesIO, cannot be used as stdin, stdout, or stderr.

Source code in process/core/protocol.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    arguments: Union[StrOrPath, Sequence[StrOrPath]],
    stdin: Optional[Union[bytes, File]] = None,
    stdout: Optional[File] = PIPE,
    stderr: Optional[File] = PIPE,
    buffer_size: Optional[int] = None,
) -> None:
    """Initialize a new `Process` instance with the given `arguments`.

    Args:
        arguments: The command and its arguments for the process.
        stdin: A [`bytes`][bytes] object containing the input data, an existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard input.
        stdout: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], and [`DEVNULL`][process.DEVNULL]) to use as the standard output.
        stderr: An existing file descriptor, an existing file object with a valid file descriptor, or special value ([`None`][None], [`PIPE`][process.PIPE], [`DEVNULL`][process.DEVNULL], and [`STDOUT`][process.STDOUT]) to use as the standard error.
        buffer_size: The buffer size for the stream operations. If `None`, the default buffer size will be used. If `0`, all operations will be unbuffered.

    Notes:
        If `arguments` is an instance of [`str`][str], it will be split into a list of arguments using [`shlex.split()`][shlex.split].

    Warning:
        A file-like object without a valid file descriptor, such as [`BytesIO`][io.BytesIO], cannot be used as `stdin`, `stdout`, or `stderr`.
    """
    ...

__del__ 

__del__() -> None

Clean up resources used by the process.

Source code in process/core/protocol.py 
70
71
72
def __del__(self) -> None:
    """Clean up resources used by the process."""
    ...

run 

run() -> Returns[Self]

Run the process.

Returns:

Type Description
Returns[Self]

The current Process instance itself.

Raises:

Type Description
ProcessAlreadyRunError

If the process has already been run.
ProcessError

If the process fails to run.
Source code in process/core/protocol.py 
146
147
148
149
150
151
152
153
154
155
156
def run(self) -> Returns[Self]:
    """Run the process.

    Returns:
        The current `Process` instance itself.

    Raises:
        ProcessAlreadyRunError: If the process has already been run.
        ProcessError: If the process fails to run.
    """
    ...

output 

output(join: bool = True) -> Returns[bytes]

Return the output from the standard output stream of the process if the process was created with stdout=PIPE.

Parameters:

Name Type Description Default
join bool

Whether to wait for the process to complete before returning the output.

 True

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stdout=PIPE.
Source code in process/core/protocol.py 
158
159
160
161
162
163
164
165
166
167
168
def output(self, join: bool = True) -> Returns[bytes]:
    """Return the output from the standard output stream of the process if the process was created with `stdout=PIPE`.

    Args:
        join: Whether to wait for the process to complete before returning the output.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessInvalidStreamError: If the process was not created with `stdout=PIPE`.
    """
    ...

error 

error(join: bool = True) -> Returns[bytes]

Return the output from the standard error stream of the process if the process was created with stderr=PIPE.

Parameters:

Name Type Description Default
join bool

Whether to wait for the process to complete before returning the output.

 True

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessInvalidStreamError

If the process was not created with stderr=PIPE.
Source code in process/core/protocol.py 
170
171
172
173
174
175
176
177
178
179
180
def error(self, join: bool = True) -> Returns[bytes]:
    """Return the output from the standard error stream of the process if the process was created with `stderr=PIPE`.

    Args:
        join: Whether to wait for the process to complete before returning the output.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessInvalidStreamError: If the process was not created with `stderr=PIPE`.
    """
    ...

join 

join(timeout: float | None = None) -> Returns[None]

Wait for the process to complete.

Parameters:

Name Type Description Default
timeout float | None

Maximum time to wait for the process to complete, in seconds. If None, wait indefinitely.

 None

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
ProcessTimeoutError

If the process does not complete within the specified timeout.
Source code in process/core/protocol.py 
182
183
184
185
186
187
188
189
190
191
192
def join(self, timeout: Optional[float] = None) -> Returns[None]:
    """Wait for the process to complete.

    Args:
        timeout: Maximum time to wait for the process to complete, in seconds. If `None`, wait indefinitely.

    Raises:
        ProcessNotRunError: If the process has not been run.
        ProcessTimeoutError: If the process does not complete within the specified timeout.
    """
    ...

signal 

signal(signal: int) -> None

Send a signal to the process.

Parameters:

Name Type Description Default
signal int

The signal number to send.

 required

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/core/protocol.py 
194
195
196
197
198
199
200
201
202
203
def signal(self, signal: int) -> None:
    """Send a signal to the process.

    Args:
        signal: The signal number to send.

    Raises:
        ProcessNotRunError: If the process has not been run.
    """
    ...

terminate 

terminate() -> None

Gracefully terminate the process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/core/protocol.py 
205
206
207
208
209
210
211
def terminate(self) -> None:
    """Gracefully terminate the process.

    Raises:
        ProcessNotRunError: If the process has not been run.
    """
    ...

kill 

kill() -> None

Forcefully kill the process.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/core/protocol.py 
213
214
215
216
217
218
219
def kill(self) -> None:
    """Forcefully kill the process.

    Raises:
        ProcessNotRunError: If the process has not been run.
    """
    ...

close 

close() -> Returns[None]

Close the process and release its resources.

Raises:

Type Description
ProcessNotRunError

If the process has not been run.
Source code in process/core/protocol.py 
221
222
223
224
225
226
def close(self) -> Returns[None]:
    """Close the process and release its resources.

    Raises:
        ProcessNotRunError: If the process has not been run.
    """

types

StrOrPath module-attribute 

StrOrPath: TypeAlias = str | PathLike

File module-attribute 

File: TypeAlias = int | IO[bytes]

⚠️ Errors

ProcessError

Base class for exceptions related to process errors.

Source code in process/core/errors.py 
1
2
class ProcessError(Exception):
    """Base class for exceptions related to process errors."""

ProcessNotRunError

Exception raised when a process has not been run.

Source code in process/core/errors.py 
5
6
class ProcessNotRunError(ProcessError):
    """Exception raised when a process has not been run."""

ProcessAlreadyRunError

Exception raised when the process has already been run.

Source code in process/core/errors.py 
 9
10
class ProcessAlreadyRunError(ProcessError):
    """Exception raised when the process has already been run."""

ProcessInvalidStreamError

Exception raised when an invalid stream is encountered in the process.

Source code in process/core/errors.py 
13
14
class ProcessInvalidStreamError(ProcessError):
    """Exception raised when an invalid stream is encountered in the process."""

ProcessTimeoutError

Exception raised when the process times out.

Source code in process/core/errors.py 
17
18
class ProcessTimeoutError(ProcessError):
    """Exception raised when the process times out."""

🔢 Constants

PIPE module-attribute 

PIPE = PIPE

Special value that can be used as the stdin, stdout, or stderr argument to indicate that a pipe to the standard stream should be opened.

STDOUT module-attribute 

STDOUT = STDOUT

Special value that can be used as the stderr argument to indicate that standard error should be redirected to standard output.

DEVNULL module-attribute 

DEVNULL = DEVNULL

Special value that can be used as the stdin, stdout, or stderr argument to indicate that the special file os.devnull should be used.

DEFAULT_BUFFER_SIZE module-attribute 

DEFAULT_BUFFER_SIZE = DEFAULT_BUFFER_SIZE

The default buffer size used by the io streams.

asyncio

DEFAULT_BUFFER_SIZE module-attribute 

DEFAULT_BUFFER_SIZE = 2 ** 16

The default buffer size used by the asyncio streams.