Python asyncio 模块，create_subprocess_shell() 实例源码

def call_method(
            self, method, arg_signature, args, return_signature, returns
    ):
        cmd = 'busctl --user -- call '
        cmd += f'{service_name} {object_path} {object_interface} {method}'
        cmd += f' "{arg_signature}"'
        for i in args:
            cmd += f' {i}'

        create = asyncio.create_subprocess_shell(
                cmd, stdout=asyncio.subprocess.PIPE
        )

        proc = await create

        # Read one line of output
        data = await proc.stdout.readline()
        line = data.decode('ascii').rstrip()

        self.assertEqual(line, f'{return_signature} {returns}')

        await proc.wait()

def process(self, event_fn):
        logger = self.get_logger()
        proc = await asyncio.create_subprocess_shell(
                self.cmd,
                stdout=asyncio.subprocess.PIPE
        )

        def strip(x):
            return x.strip()
        stdout, stderr = await proc.communicate()
        content = stdout.decode().strip()
        transform = partial(self.transform_fn, logger=logger, **self.args)
        metrics = filter(
                is_parsed,
                map(transform, map(strip, content.split("\n")))
        )

        for key, val in metrics:
            event_fn(service=self.prefix + key,
                     state="ok",
                     metric_f=float(val))

def cat(loop):

    #
    # ????:
    #   - ????:
    #
    proc = yield from asyncio.create_subprocess_shell("cat",
                                                      stdin=PIPE,
                                                      stdout=PIPE)
    print("pid: %s" % proc.pid)

    message = "Hello World!"
    print("cat write: %r" % message)

    stdout, stderr = yield from proc.communicate(message.encode('ascii'))
    print("cat read: %r" % stdout.decode('ascii'))

    exitcode = yield from proc.wait()
    print("(exit code %s)" % exitcode)

def run_cmd(self, *args, **kwargs):
        if len(args) > 1:
            value = args
        elif args:
            value = args[0]
        elif kwargs:
            value = kwargs
        else:
            value = None
        cmd = self.make_command(value)
        self.logger.info(' '.join(cmd))
        if len(cmd) > 1:
            coro = asyncio.create_subprocess_exec
        else:
            coro = asyncio.create_subprocess_shell
        self._process = await coro(
            *cmd, stdout=self._stdout, stderr=self._stderr, loop=self.loop)
        if self._wait:
            await self._process.wait()
        if self._stdout:
            data = await self._process.stdout.read()
            return self.decode(data)

def shell(cmd: str) -> str:
    """
    Executes a shell command asynchronously.

    Parameters
    ----------
    cmd
        The command to execute.

    Returns
    -------
        The command output.
    """
    process = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    results = await process.communicate()
    return ''.join(x.decode() for x in results)

def ci(ref, targets=None, all=False):
    targets = ' '.join('-t %s' % t for t in targets or [] if t)
    cmd = (
        'cd {root} && '
        'FIRE_UID={uid} ./fire ci {ref.scope.name} {ref.val} {all} {targets}'
        .format(
            root=root,
            ref=ref,
            uid=str(uuid.uuid4().hex[:8]),
            targets=targets,
            all=(all and '--all' or '')
        )
    )
    log.info(cmd)
    proc = await asyncio.create_subprocess_shell(cmd, executable='/bin/bash')
    code = await proc.wait()
    log.info('code=%s: %s', code, cmd)
    return code

def start(self, cmd, *cmd_args):
        """
        """
        command = cmd

        # Make sure the provided arguments are safe:
        if cmd_args:
            args = __class__.escape_args(*cmd_args)
            command = "{} {}".format(command, args)

        # And then launch the command:
        proc = await asyncio.create_subprocess_shell(command, stderr=PIPE)
        stdout_data, stderr_data = await proc.communicate()

        if stdout_data:
            print(stdout_data)

        if stderr_data:
            self.handle_error(stderr_data)

        # Shell commands are supposed to return 0 on success.
        # When an error occurs, the cmd is supposed to print a message
        # on stderr. This message is caught and passed to
        # the `handle_error` method.
        return True if proc.returncode is 0 else False

def async_play(self):
        _LOGGER.warning('Trying to play %s', self._url)
        cmd = self._build_cmd(add_account=False)
        _LOGGER.info('cmd: %s', cmd)
        proc = yield from asyncio.create_subprocess_shell(
            cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE)
        (stdout, stderr) = yield from proc.communicate()

        if proc.returncode == 0 and len(stderr) == 0:
            url = stdout.decode('utf-8').strip('\n')
            if url.startswith('https://'):
                return (yield from self.async_play_url(url))
            else:
                _LOGGER.warning('Got invalid response: %s', url)
        elif self._live:
            return (yield from self.async_live_stream(stderr.decode('utf-8')))

        return False

def run(self, command):
        try:
            process = await asyncio.create_subprocess_shell(
                command, stdout = PIPE, stderr = PIPE)
            # xxx big room for improvement: show stuff on the fly
            # and not in a big batch at the end
            stdout, stderr = await process.communicate()
            # xxx unsure what datatype should be
            # except that it's not EXTENDED_DATA_STDERR
            self.lines(stdout, 0)
            self.lines(stderr, EXTENDED_DATA_STDERR)
            retcod = await process.wait()
            return retcod
        except Exception as e:
            line = "LocalNode: Could not run local command {} - {}"\
                   .format(command, e)
            self.formatter.line(line, EXTENDED_DATA_STDERR, self.hostname)

def process(self, event_fn):
        proc = await asyncio.create_subprocess_shell(self.script)
        exitcode = await proc.wait()
        state = "ok" if exitcode == 0 else "failure"
        event_fn(service=self.prefix + "shell",
                 state=state,
                 metric_f=1.0,
                 description="Exit code: {0}".format(exitcode)
                 )

def run(self):
        try:
            fixed_command = self._fix_command(self.command)
            self._process = await asyncio.create_subprocess_shell(
                fixed_command, stdout=asyncio.subprocess.PIPE)
        except Exception as ex:
            self._error     = ex
            self._pass_rate = 'N/A'
            self._status    = 'failed'

def start(cmd, input=None, **kwds):
    kwds['stdout'] = PIPE
    kwds['stderr'] = PIPE
    if input is None and 'stdin' not in kwds:
        kwds['stdin'] = None
    else:
        kwds['stdin'] = PIPE
    proc = yield from asyncio.create_subprocess_shell(cmd, **kwds)

    tasks = []
    if input is not None:
        tasks.append(send_input(proc.stdin, input))
    else:
        print('No stdin')
    if proc.stderr is not None:
        tasks.append(log_errors(proc.stderr))
    else:
        print('No stderr')
    if proc.stdout is not None:
        tasks.append(read_stdout(proc.stdout))
    else:
        print('No stdout')

    if tasks:
        # feed stdin while consuming stdout to avoid hang
        # when stdin pipe is full
        yield from asyncio.wait(tasks)

    exitcode = yield from proc.wait()
    print("exit code: %s" % exitcode)

def test_shell(self):
        create = asyncio.create_subprocess_shell('exit 7',
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 7)

def test_start_new_session(self):
        # start the new process in a new session
        create = asyncio.create_subprocess_shell('exit 8',
                                                 start_new_session=True,
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 8)

def cat(loop):
    proc = await asyncio.create_subprocess_shell('ping', stdin=PIPE, stdout=PIPE)

    print("pid: %s" % proc.pid)

    message = "Hello World!"

    print("cat read: %r" % message)

    stdout, stderr = await proc.communicate(message.encode('ascii'))
    print("cat read: %r" % stdout.decode('ascii'))

    exitcode = await proc.wait()
    print("exit code %s" % exitcode)

def run_subprocess(cmd, loop=None):
    loop = loop or asyncio.get_event_loop()
    try:
        proc = await asyncio.create_subprocess_shell(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )
    # XXX: On platforms that have event loops that don't support subprocesses,
    #      the NotImplementedError exception will always be tripped. From what
    #      I've heard always raising an exception and triggering the except
    #      statement it up to 10-20x slower. But there's no LBYL or one-shot
    #      way of checking whether or not a given even loop supports subprocesses.
    except NotImplementedError:
        # The default event loop for Windows doesn't support subprocesses. To
        # make matters worse, we can't use the proactor event loop because
        # there is a regression in Python 3.6 where an AssertionError is
        # thrown when reading the data.
        # 
        # This is why we have no choice but to resort to using an executor.
        # If anyone can supply a minimal repro of this issue and submit it
        # to the Python bug tracker that would be really nice (I can't sadly
        # because I'm not smart enough to figure out what's going on).
        #
        # See: https://github.com/Rapptz/discord.py/issues/859
        with subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
        ) as proc:
            try:
                result = await loop.run_in_executor(None, proc.communicate)
            except:
                def kill():
                    proc.kill()
                    proc.wait()
                # Wait for the process to die but don't block the loop.
                await loop.run_in_executor(None, kill)
                raise
    else:
        result = await proc.communicate()

    return [x.decode('utf-8') for x in result]

def test_shell(self):
        create = asyncio.create_subprocess_shell('exit 7',
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 7)

def test_start_new_session(self):
        # start the new process in a new session
        create = asyncio.create_subprocess_shell('exit 8',
                                                 start_new_session=True,
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 8)

def shell(command: str):
    process = await asyncio.create_subprocess_shell(
        command,
        stderr=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
    )

    out, err = map(lambda s: s.decode('utf-8'), await process.communicate())
    return f'{out}{err}'.strip()

def start(self, path, game):
        info("chat %s: frob start", self._chat_id)
        self._process = await asyncio.create_subprocess_shell(
            'frob -iplain {}/{}.gam'.format(path, game),
            stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.STDOUT, preexec_fn=os.setpgrp)

        if os.path.exists(path + '/last.sav'):
            await self._read_output()  # just ignore all previous output
            self.restore_game('last')
        else:
            self._messages_to_skip = 1  # ignore frobTADS intro msg

def test_shell(self):
        create = asyncio.create_subprocess_shell('exit 7',
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 7)

def test_start_new_session(self):
        # start the new process in a new session
        create = asyncio.create_subprocess_shell('exit 8',
                                                 start_new_session=True,
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 8)

def run_subprocess(cmd, loop=None):
    try:
        proc = await asyncio.create_subprocess_shell(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        res = await proc.communicate()
    except NotImplementedError:
        loop = loop or asyncio.get_event_loop()
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
        res = await loop.run_in_executor(None, proc.communicate)
    return [s.decode('utf8') for s in res]

def make_async_call(cmd, stderr, stdout):
    return asyncio.create_subprocess_shell(cmd, stdout=stdout, stderr=stderr)

def run_script():
    process = await asyncio.create_subprocess_shell(
        'python3',
        stdout=asyncio.subprocess.PIPE,
        stdin=asyncio.subprocess.PIPE,
    )

    # Write a simple Python script to the interpreter
    process.stdin.write(b'\n'.join((
        b'import math',
        b'x = 2 ** 8',
        b'y = math.sqrt(x)',
        b'z = math.sqrt(y)',
        b'print("x: %d" % x)',
        b'print("y: %d" % y)',
        b'print("z: %d" % z)',
        b'for i in range(int(z)):',
        b'    print("i: %d" % i)',
    )))
    # Make sure the stdin is flushed asynchronously
    await process.stdin.drain()
    # And send the end of file so the Python interpreter will
    # start processing the input. Without this the process will
    # stall forever.
    process.stdin.write_eof()

    # Fetch the lines from the stdout asynchronously
    async for out in process.stdout:
        # Decode the output from bytes and strip the whitespace
        # (newline) at the right
        print(out.decode('utf-8').rstrip())

    # Wait for the process to exit
    await process.wait()

def run_cmd_async(cmd, cwd, env=None, fail=True, shell=False, liveupdate=True):
    """
        Run a command asynchronously.
    """
    # pylint: disable=too-many-arguments

    env = env or {}
    cmdstr = cmd
    if not shell:
        cmdstr = ' '.join(cmd)
    logging.info('%s$ %s', cwd, cmdstr)

    logo = LogOutput(liveupdate)

    if shell:
        process = yield from asyncio.create_subprocess_shell(
            cmd,
            env=env,
            cwd=cwd,
            universal_newlines=True,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE)
    else:
        process = yield from asyncio.create_subprocess_exec(
            *cmd,
            cwd=cwd,
            env=env,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE)

    yield from asyncio.wait([
        _read_stream(process.stdout, logo.log_stdout),
        _read_stream(process.stderr, logo.log_stderr)
    ])
    ret = yield from process.wait()

    if ret and fail:
        msg = 'Command "{cwd}$ {cmd}" failed'.format(cwd=cwd, cmd=cmdstr)
        if logo.stderr:
            msg += '\n--- Error summary ---\n'
            for line in logo.stderr:
                msg += line
        logging.error(msg)

    return (ret, ''.join(logo.stdout))

def start(self) -> None:
        if self.proc is not None:
            raise RuntimeError("process already running")
        kwargs = {}  # type: Dict[str, Any]
        if isinstance(self.config.command, list):
            create = asyncio.create_subprocess_exec
            cmd = self.config.command
        else:
            if self.config.shell:
                create = asyncio.create_subprocess_exec
                cmd = [self.config.shell, '-c', self.config.command]
            else:
                create = asyncio.create_subprocess_shell
                cmd = [self.config.command]
        if self.config.environment:
            env = dict(os.environ)
            for envvar in self.config.environment:
                env[envvar['key']] = envvar['value']
                self.env = env
            kwargs['env'] = env
        logger.debug("%s: will execute argv %r", self.config.name, cmd)
        if self.config.captureStderr:
            kwargs['stderr'] = asyncio.subprocess.PIPE
        if self.config.captureStdout:
            kwargs['stdout'] = asyncio.subprocess.PIPE
        if self.config.executionTimeout:
            self.execution_deadline = (time.perf_counter() +
                                       self.config.executionTimeout)

        self.proc = await create(*cmd, **kwargs)

        await self._on_start()

        if self.config.captureStderr:
            assert self.proc.stderr is not None
            self._stderr_reader = \
                StreamReader(self.config.name, 'stderr', self.proc.stderr,
                             self.config.saveLimit)
        if self.config.captureStdout:
            assert self.proc.stdout is not None
            self._stdout_reader = \
                StreamReader(self.config.name, 'stdout', self.proc.stdout,
                             self.config.saveLimit)