如何编写适配asyncio的subprocess.check_output异步实现？

可以基于asyncio.subprocess模块实现完全兼容原生接口的异步版本，无需修改原有调用逻辑即可直接替换：

完整实现代码 #

import asyncio
from subprocess import CalledProcessError, TimeoutExpired
from typing import Union, Sequence, Optional


async def async_check_output(
    args: Union[str, Sequence[str]],
    *,
    stdin: Optional[Union[bytes, str]] = None,
    stderr=None,
    shell: bool = False,
    cwd: Optional[str] = None,
    env: Optional[dict] = None,
    universal_newlines: bool = False,
    text: Optional[bool] = None,
    encoding: Optional[str] = None,
    errors: Optional[str] = None,
    timeout: Optional[float] = None
) -> Union[bytes, str]:
    # 对齐原生check_output的text参数处理逻辑
    text = text if text is not None else universal_newlines
    
    # 创建异步子进程
    if shell:
        proc = await asyncio.create_subprocess_shell(
            args,
            stdin=asyncio.subprocess.PIPE if stdin is not None else None,
            stdout=asyncio.subprocess.PIPE,
            stderr=stderr,
            cwd=cwd,
            env=env,
            text=text,
            encoding=encoding,
            errors=errors
        )
    else:
        proc = await asyncio.create_subprocess_exec(
            *args,
            stdin=asyncio.subprocess.PIPE if stdin is not None else None,
            stdout=asyncio.subprocess.PIPE,
            stderr=stderr,
            cwd=cwd,
            env=env,
            text=text,
            encoding=encoding,
            errors=errors
        )

    try:
        # 执行进程并等待返回，设置超时
        stdout, stderr_val = await asyncio.wait_for(
            proc.communicate(input=stdin),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        proc.kill()
        await proc.wait()
        raise TimeoutExpired(args, timeout) from None

    # 非0返回码抛出异常，对齐原生行为
    if proc.returncode != 0:
        raise CalledProcessError(
            proc.returncode,
            args,
            output=stdout,
            stderr=stderr_val
        )

    return stdout

使用示例 #

和原生subprocess.check_output用法完全一致，只需要加上await关键字即可：

# 同步调用原写法
# import subprocess
# res = subprocess.check_output(["echo", "hello test"], text=True, timeout=5)

# 异步替换后写法
res = await async_check_output(["echo", "hello test"], text=True, timeout=5)
print(res) # 输出 hello test

行为对齐说明 #

• 进程返回非0状态码时自动抛出CalledProcessError，异常属性和原生完全一致
• 超时场景自动杀死进程并抛出TimeoutExpired
• 支持所有原生参数：shell、cwd、env、text、编码配置等都可以直接沿用原有传参
• 需要合并stderr到输出时，和原生一样传入stderr=subprocess.STDOUT即可

内容的提问来源于stack exchange，提问作者Zachary Vance