Coverage for Lib/asyncio/subprocess.py: 83%
179 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 01:31 +0000
1__all__ = 'create_subprocess_exec', 'create_subprocess_shell'
3import subprocess
5from . import events
6from . import protocols
7from . import streams
8from . import tasks
9from .log import logger
12PIPE = subprocess.PIPE
13STDOUT = subprocess.STDOUT
14DEVNULL = subprocess.DEVNULL
17class SubprocessStreamProtocol(streams.FlowControlMixin,
18 protocols.SubprocessProtocol):
19 """Like StreamReaderProtocol, but for a subprocess."""
21 def __init__(self, limit, loop):
22 super().__init__(loop=loop)
23 self._limit = limit
24 self.stdin = self.stdout = self.stderr = None
25 self._transport = None
26 self._process_exited = False
27 self._pipe_fds = []
28 self._stdin_closed = self._loop.create_future()
30 def __repr__(self):
31 info = [self.__class__.__name__]
32 if self.stdin is not None:
33 info.append(f'stdin={self.stdin!r}')
34 if self.stdout is not None:
35 info.append(f'stdout={self.stdout!r}')
36 if self.stderr is not None:
37 info.append(f'stderr={self.stderr!r}')
38 return '<{}>'.format(' '.join(info))
40 def connection_made(self, transport):
41 self._transport = transport
43 stdout_transport = transport.get_pipe_transport(1)
44 if stdout_transport is not None:
45 self.stdout = streams.StreamReader(limit=self._limit,
46 loop=self._loop)
47 self.stdout.set_transport(stdout_transport)
48 self._pipe_fds.append(1)
50 stderr_transport = transport.get_pipe_transport(2)
51 if stderr_transport is not None:
52 self.stderr = streams.StreamReader(limit=self._limit,
53 loop=self._loop)
54 self.stderr.set_transport(stderr_transport)
55 self._pipe_fds.append(2)
57 stdin_transport = transport.get_pipe_transport(0)
58 if stdin_transport is not None:
59 self.stdin = streams.StreamWriter(stdin_transport,
60 protocol=self,
61 reader=None,
62 loop=self._loop)
64 def pipe_data_received(self, fd, data):
65 if fd == 1:
66 reader = self.stdout
67 elif fd == 2: 67 ↛ 70line 67 didn't jump to line 70 because the condition on line 67 was always true
68 reader = self.stderr
69 else:
70 reader = None
71 if reader is not None: 71 ↛ exitline 71 didn't return from function 'pipe_data_received' because the condition on line 71 was always true
72 reader.feed_data(data)
74 def pipe_connection_lost(self, fd, exc):
75 if fd == 0:
76 pipe = self.stdin
77 if pipe is not None: 77 ↛ 79line 77 didn't jump to line 79 because the condition on line 77 was always true
78 pipe.close()
79 self.connection_lost(exc)
80 if exc is None:
81 self._stdin_closed.set_result(None)
82 else:
83 self._stdin_closed.set_exception(exc)
84 # Since calling `wait_closed()` is not mandatory,
85 # we shouldn't log the traceback if this is not awaited.
86 self._stdin_closed._log_traceback = False
87 return
88 if fd == 1:
89 reader = self.stdout
90 elif fd == 2: 90 ↛ 93line 90 didn't jump to line 93 because the condition on line 90 was always true
91 reader = self.stderr
92 else:
93 reader = None
94 if reader is not None: 94 ↛ 100line 94 didn't jump to line 100 because the condition on line 94 was always true
95 if exc is None: 95 ↛ 98line 95 didn't jump to line 98 because the condition on line 95 was always true
96 reader.feed_eof()
97 else:
98 reader.set_exception(exc)
100 if fd in self._pipe_fds: 100 ↛ 102line 100 didn't jump to line 102 because the condition on line 100 was always true
101 self._pipe_fds.remove(fd)
102 self._maybe_close_transport()
104 def process_exited(self):
105 self._process_exited = True
106 self._maybe_close_transport()
108 def _maybe_close_transport(self):
109 if len(self._pipe_fds) == 0 and self._process_exited:
110 self._transport.close()
111 self._transport = None
113 def _get_close_waiter(self, stream):
114 if stream is self.stdin:
115 return self._stdin_closed
118class Process:
119 def __init__(self, transport, protocol, loop):
120 self._transport = transport
121 self._protocol = protocol
122 self._loop = loop
123 self.stdin = protocol.stdin
124 self.stdout = protocol.stdout
125 self.stderr = protocol.stderr
126 self.pid = transport.get_pid()
127 self._communication_started = False
128 self._input = None
129 self._input_written = False
130 self._stdout_buf = bytearray()
131 self._stderr_buf = bytearray()
133 def __repr__(self):
134 return f'<{self.__class__.__name__} {self.pid}>'
136 @property
137 def returncode(self):
138 return self._transport.get_returncode()
140 async def wait(self):
141 """Wait until the process exit and return the process return code."""
142 return await self._transport._wait()
144 def send_signal(self, signal):
145 self._transport.send_signal(signal)
147 def terminate(self):
148 self._transport.terminate()
150 def kill(self):
151 self._transport.kill()
153 async def _feed_stdin(self, input):
154 debug = self._loop.get_debug()
155 try:
156 if input is not None and not self._input_written:
157 self.stdin.write(input)
158 self._input_written = True
159 if debug: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true
160 logger.debug(
161 '%r communicate: feed stdin (%s bytes)', self, len(input))
163 await self.stdin.drain()
164 except (BrokenPipeError, ConnectionResetError) as exc:
165 # communicate() ignores BrokenPipeError and ConnectionResetError.
166 # write() and drain() can raise these exceptions.
167 if debug: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true
168 logger.debug('%r communicate: stdin got %r', self, exc)
170 if debug: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 logger.debug('%r communicate: close stdin', self)
172 self.stdin.close()
174 async def _noop(self):
175 return None
177 async def _read_stream(self, fd):
178 transport = self._transport.get_pipe_transport(fd)
179 if fd == 2:
180 stream = self.stderr
181 buf = self._stderr_buf
182 else:
183 assert fd == 1
184 stream = self.stdout
185 buf = self._stdout_buf
186 if self._loop.get_debug(): 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true
187 name = 'stdout' if fd == 1 else 'stderr'
188 logger.debug('%r communicate: read %s', self, name)
189 # Append each block to the persistent buffer as soon as it is
190 # read so that no output is lost if this coroutine is cancelled.
191 while block := await stream.read(stream._limit):
192 buf += block
193 if self._loop.get_debug(): 193 ↛ 194line 193 didn't jump to line 194 because the condition on line 193 was never true
194 name = 'stdout' if fd == 1 else 'stderr'
195 logger.debug('%r communicate: close %s', self, name)
196 transport.close()
198 async def communicate(self, input=None):
199 if self._communication_started:
200 if input:
201 raise ValueError(
202 "Cannot send input after starting communication")
203 else:
204 self._input = input
205 self._communication_started = True
206 if self.stdin is not None:
207 stdin = self._feed_stdin(self._input)
208 else:
209 stdin = self._noop()
210 if self.stdout is not None:
211 stdout = self._read_stream(1)
212 else:
213 stdout = self._noop()
214 if self.stderr is not None:
215 stderr = self._read_stream(2)
216 else:
217 stderr = self._noop()
218 await tasks.gather(stdin, stdout, stderr)
219 await self.wait()
220 if self.stdout is not None:
221 stdout = self._stdout_buf.take_bytes()
222 else:
223 stdout = None
224 if self.stderr is not None:
225 stderr = self._stderr_buf.take_bytes()
226 else:
227 stderr = None
228 return (stdout, stderr)
231async def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,
232 limit=streams._DEFAULT_LIMIT, **kwds):
233 loop = events.get_running_loop()
234 protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
235 loop=loop)
236 transport, protocol = await loop.subprocess_shell(
237 protocol_factory,
238 cmd, stdin=stdin, stdout=stdout,
239 stderr=stderr, **kwds)
240 return Process(transport, protocol, loop)
243async def create_subprocess_exec(program, *args, stdin=None, stdout=None,
244 stderr=None, limit=streams._DEFAULT_LIMIT,
245 **kwds):
246 loop = events.get_running_loop()
247 protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
248 loop=loop)
249 transport, protocol = await loop.subprocess_exec(
250 protocol_factory,
251 program, *args,
252 stdin=stdin, stdout=stdout,
253 stderr=stderr, **kwds)
254 return Process(transport, protocol, loop)