__all__ = ['create_subprocess_exec', 'create_subprocess_shell']

[0] Fix | Delete

[1] Fix | Delete

import subprocess

[2] Fix | Delete

[3] Fix | Delete

from . import events

[4] Fix | Delete

from . import protocols

[5] Fix | Delete

from . import streams

[6] Fix | Delete

from . import tasks

[7] Fix | Delete

from .coroutines import coroutine

[8] Fix | Delete

from .log import logger

[9] Fix | Delete

[10] Fix | Delete

[11] Fix | Delete

PIPE = subprocess.PIPE

[12] Fix | Delete

STDOUT = subprocess.STDOUT

[13] Fix | Delete

DEVNULL = subprocess.DEVNULL

[14] Fix | Delete

[15] Fix | Delete

[16] Fix | Delete

class SubprocessStreamProtocol(streams.FlowControlMixin,

[17] Fix | Delete

protocols.SubprocessProtocol):

[18] Fix | Delete

"""Like StreamReaderProtocol, but for a subprocess."""

[19] Fix | Delete

[20] Fix | Delete

def __init__(self, limit, loop):

[21] Fix | Delete

super().__init__(loop=loop)

[22] Fix | Delete

self._limit = limit

[23] Fix | Delete

self.stdin = self.stdout = self.stderr = None

[24] Fix | Delete

self._transport = None

[25] Fix | Delete

self._process_exited = False

[26] Fix | Delete

self._pipe_fds = []

[27] Fix | Delete

[28] Fix | Delete

def __repr__(self):

[29] Fix | Delete

info = [self.__class__.__name__]

[30] Fix | Delete

if self.stdin is not None:

[31] Fix | Delete

info.append('stdin=%r' % self.stdin)

[32] Fix | Delete

if self.stdout is not None:

[33] Fix | Delete

info.append('stdout=%r' % self.stdout)

[34] Fix | Delete

if self.stderr is not None:

[35] Fix | Delete

info.append('stderr=%r' % self.stderr)

[36] Fix | Delete

return '<%s>' % ' '.join(info)

[37] Fix | Delete

[38] Fix | Delete

def connection_made(self, transport):

[39] Fix | Delete

self._transport = transport

[40] Fix | Delete

[41] Fix | Delete

stdout_transport = transport.get_pipe_transport(1)

[42] Fix | Delete

if stdout_transport is not None:

[43] Fix | Delete

self.stdout = streams.StreamReader(limit=self._limit,

[44] Fix | Delete

loop=self._loop)

[45] Fix | Delete

self.stdout.set_transport(stdout_transport)

[46] Fix | Delete

self._pipe_fds.append(1)

[47] Fix | Delete

[48] Fix | Delete

stderr_transport = transport.get_pipe_transport(2)

[49] Fix | Delete

if stderr_transport is not None:

[50] Fix | Delete

self.stderr = streams.StreamReader(limit=self._limit,

[51] Fix | Delete

loop=self._loop)

[52] Fix | Delete

self.stderr.set_transport(stderr_transport)

[53] Fix | Delete

self._pipe_fds.append(2)

[54] Fix | Delete

[55] Fix | Delete

stdin_transport = transport.get_pipe_transport(0)

[56] Fix | Delete

if stdin_transport is not None:

[57] Fix | Delete

self.stdin = streams.StreamWriter(stdin_transport,

[58] Fix | Delete

protocol=self,

[59] Fix | Delete

reader=None,

[60] Fix | Delete

loop=self._loop)

[61] Fix | Delete

[62] Fix | Delete

def pipe_data_received(self, fd, data):

[63] Fix | Delete

if fd == 1:

[64] Fix | Delete

reader = self.stdout

[65] Fix | Delete

elif fd == 2:

[66] Fix | Delete

reader = self.stderr

[67] Fix | Delete

else:

[68] Fix | Delete

reader = None

[69] Fix | Delete

if reader is not None:

[70] Fix | Delete

reader.feed_data(data)

[71] Fix | Delete

[72] Fix | Delete

def pipe_connection_lost(self, fd, exc):

[73] Fix | Delete

if fd == 0:

[74] Fix | Delete

pipe = self.stdin

[75] Fix | Delete

if pipe is not None:

[76] Fix | Delete

pipe.close()

[77] Fix | Delete

self.connection_lost(exc)

[78] Fix | Delete

return

[79] Fix | Delete

if fd == 1:

[80] Fix | Delete

reader = self.stdout

[81] Fix | Delete

elif fd == 2:

[82] Fix | Delete

reader = self.stderr

[83] Fix | Delete

else:

[84] Fix | Delete

reader = None

[85] Fix | Delete

if reader != None:

[86] Fix | Delete

if exc is None:

[87] Fix | Delete

reader.feed_eof()

[88] Fix | Delete

else:

[89] Fix | Delete

reader.set_exception(exc)

[90] Fix | Delete

[91] Fix | Delete

if fd in self._pipe_fds:

[92] Fix | Delete

self._pipe_fds.remove(fd)

[93] Fix | Delete

self._maybe_close_transport()

[94] Fix | Delete

[95] Fix | Delete

def process_exited(self):

[96] Fix | Delete

self._process_exited = True

[97] Fix | Delete

self._maybe_close_transport()

[98] Fix | Delete

[99] Fix | Delete

def _maybe_close_transport(self):

[100] Fix | Delete

if len(self._pipe_fds) == 0 and self._process_exited:

[101] Fix | Delete

self._transport.close()

[102] Fix | Delete

self._transport = None

[103] Fix | Delete

[104] Fix | Delete

[105] Fix | Delete

class Process:

[106] Fix | Delete

def __init__(self, transport, protocol, loop):

[107] Fix | Delete

self._transport = transport

[108] Fix | Delete

self._protocol = protocol

[109] Fix | Delete

self._loop = loop

[110] Fix | Delete

self.stdin = protocol.stdin

[111] Fix | Delete

self.stdout = protocol.stdout

[112] Fix | Delete

self.stderr = protocol.stderr

[113] Fix | Delete

self.pid = transport.get_pid()

[114] Fix | Delete

[115] Fix | Delete

def __repr__(self):

[116] Fix | Delete

return '<%s %s>' % (self.__class__.__name__, self.pid)

[117] Fix | Delete

[118] Fix | Delete

@property

[119] Fix | Delete

def returncode(self):

[120] Fix | Delete

return self._transport.get_returncode()

[121] Fix | Delete

[122] Fix | Delete

@coroutine

[123] Fix | Delete

def wait(self):

[124] Fix | Delete

"""Wait until the process exit and return the process return code.

[125] Fix | Delete

[126] Fix | Delete

This method is a coroutine."""

[127] Fix | Delete

return (yield from self._transport._wait())

[128] Fix | Delete

[129] Fix | Delete

def send_signal(self, signal):

[130] Fix | Delete

self._transport.send_signal(signal)

[131] Fix | Delete

[132] Fix | Delete

def terminate(self):

[133] Fix | Delete

self._transport.terminate()

[134] Fix | Delete

[135] Fix | Delete

def kill(self):

[136] Fix | Delete

self._transport.kill()

[137] Fix | Delete

[138] Fix | Delete

@coroutine

[139] Fix | Delete

def _feed_stdin(self, input):

[140] Fix | Delete

debug = self._loop.get_debug()

[141] Fix | Delete

self.stdin.write(input)

[142] Fix | Delete

if debug:

[143] Fix | Delete

logger.debug('%r communicate: feed stdin (%s bytes)',

[144] Fix | Delete

self, len(input))

[145] Fix | Delete

try:

[146] Fix | Delete

yield from self.stdin.drain()

[147] Fix | Delete

except (BrokenPipeError, ConnectionResetError) as exc:

[148] Fix | Delete

# communicate() ignores BrokenPipeError and ConnectionResetError

[149] Fix | Delete

if debug:

[150] Fix | Delete

logger.debug('%r communicate: stdin got %r', self, exc)

[151] Fix | Delete

[152] Fix | Delete

if debug:

[153] Fix | Delete

logger.debug('%r communicate: close stdin', self)

[154] Fix | Delete

self.stdin.close()

[155] Fix | Delete

[156] Fix | Delete

@coroutine

[157] Fix | Delete

def _noop(self):

[158] Fix | Delete

return None

[159] Fix | Delete

[160] Fix | Delete

@coroutine

[161] Fix | Delete

def _read_stream(self, fd):

[162] Fix | Delete

transport = self._transport.get_pipe_transport(fd)

[163] Fix | Delete

if fd == 2:

[164] Fix | Delete

stream = self.stderr

[165] Fix | Delete

else:

[166] Fix | Delete

assert fd == 1

[167] Fix | Delete

stream = self.stdout

[168] Fix | Delete

if self._loop.get_debug():

[169] Fix | Delete

name = 'stdout' if fd == 1 else 'stderr'

[170] Fix | Delete

logger.debug('%r communicate: read %s', self, name)

[171] Fix | Delete

output = yield from stream.read()

[172] Fix | Delete

if self._loop.get_debug():

[173] Fix | Delete

name = 'stdout' if fd == 1 else 'stderr'

[174] Fix | Delete

logger.debug('%r communicate: close %s', self, name)

[175] Fix | Delete

transport.close()

[176] Fix | Delete

return output

[177] Fix | Delete

[178] Fix | Delete

@coroutine

[179] Fix | Delete

def communicate(self, input=None):

[180] Fix | Delete

if input is not None:

[181] Fix | Delete

stdin = self._feed_stdin(input)

[182] Fix | Delete

else:

[183] Fix | Delete

stdin = self._noop()

[184] Fix | Delete

if self.stdout is not None:

[185] Fix | Delete

stdout = self._read_stream(1)

[186] Fix | Delete

else:

[187] Fix | Delete

stdout = self._noop()

[188] Fix | Delete

if self.stderr is not None:

[189] Fix | Delete

stderr = self._read_stream(2)

[190] Fix | Delete

else:

[191] Fix | Delete

stderr = self._noop()

[192] Fix | Delete

stdin, stdout, stderr = yield from tasks.gather(stdin, stdout, stderr,

[193] Fix | Delete

loop=self._loop)

[194] Fix | Delete

yield from self.wait()

[195] Fix | Delete

return (stdout, stderr)

[196] Fix | Delete

[197] Fix | Delete

[198] Fix | Delete

@coroutine

[199] Fix | Delete

def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,

[200] Fix | Delete

loop=None, limit=streams._DEFAULT_LIMIT, **kwds):

[201] Fix | Delete

if loop is None:

[202] Fix | Delete

loop = events.get_event_loop()

[203] Fix | Delete

protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,

[204] Fix | Delete

loop=loop)

[205] Fix | Delete

transport, protocol = yield from loop.subprocess_shell(

[206] Fix | Delete

protocol_factory,

[207] Fix | Delete

cmd, stdin=stdin, stdout=stdout,

[208] Fix | Delete

stderr=stderr, **kwds)

[209] Fix | Delete

return Process(transport, protocol, loop)

[210] Fix | Delete

[211] Fix | Delete

@coroutine

[212] Fix | Delete

def create_subprocess_exec(program, *args, stdin=None, stdout=None,

[213] Fix | Delete

stderr=None, loop=None,

[214] Fix | Delete

limit=streams._DEFAULT_LIMIT, **kwds):

[215] Fix | Delete

if loop is None:

[216] Fix | Delete

loop = events.get_event_loop()

[217] Fix | Delete

protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,

[218] Fix | Delete

loop=loop)

[219] Fix | Delete

transport, protocol = yield from loop.subprocess_exec(

[220] Fix | Delete

protocol_factory,

[221] Fix | Delete

program, *args,

[222] Fix | Delete

stdin=stdin, stdout=stdout,

[223] Fix | Delete

stderr=stderr, **kwds)

[224] Fix | Delete

return Process(transport, protocol, loop)

[225] Fix | Delete

[226] Fix | Delete