Можно ли использовать подпроцесс asyncio с contextmanager?

Asyncio обеспечивает параллельное выполнение за счет приостановки всего, что может заблокировать. Чтобы это работало, весь код должен быть внутри обратных вызовов или сопрограмм и воздержитесь от вызова блокирующих функций, таких как time.sleep(). Помимо этого, в вашем коде есть некоторые другие проблемы, например, await asyncio.wait([x]) эквивалентно await x, а это означает, что open_subprocess не будет выполняться до тех пор, пока не будет выполнено все чтение потока.

Правильный способ структурировать код — переместить код верхнего уровня в async def и использовать диспетчер асинхронного контекста. Например:

import asyncio
import contextlib

async def read_stream(proc, stream, callback):
    while proc.returncode is None:
        data = await stream.readline()
        if data:
            callback(data.decode().rstrip())
        else:
            break

@contextlib.asynccontextmanager
async def open_subprocess(cmd, *args, stdout_callback=print):
    proc = await asyncio.create_subprocess_exec(
        cmd, *args, stdout=asyncio.subprocess.PIPE)
    asyncio.create_task(read_stream(proc, proc.stdout, stdout_callback))
    yield proc
    if proc.returncode is None:
        proc.terminate()
        await proc.wait()

async def main():
    def stdout_callback(data):
        print('STDOUT:', data)

    async with open_subprocess('ping', '-c', '4', 'localhost',
                               stdout_callback=stdout_callback) as proc:
        for i in range(2):
            print(f'RUNNING SUBPROCESS {proc.pid}...')
            await asyncio.sleep(1)

    print(f'RETURN CODE: {proc.returncode}')

asyncio.run(main())

Если вы настаиваете на смешивании синхронизирующего и асинхронного кода, вам необходимо полностью разделить их, запустив цикл обработки событий asyncio в отдельном потоке. Тогда ваш основной поток не сможет напрямую обращаться к асинхронным объектам, таким как proc, потому что они не являются потокобезопасными. Вам нужно последовательно использовать call_soon_threadsafe и run_coroutine_threadsafe для связи с циклом событий.

Этот подход сложен и требует взаимодействия между потоками и работы с циклами событий, поэтому я бы не рекомендовал его, кроме как в качестве учебного упражнения. Не говоря уже о том, что если вы используете другой поток, вам вообще не нужно возиться с asyncio — вы можете напрямую выполнять вызовы синхронизации в другом потоке. Но, сказав это, вот возможная реализация:

import asyncio
import contextlib
import concurrent.futures
import threading

async def read_stream(proc, stream, callback):
    while proc.returncode is None:
        data = await stream.readline()
        if data:
            callback(data.decode().rstrip())
        else:
            break

async def stream_subprocess(cmd, *args, proc_data_future, stdout_callback=print):
    try:
        proc = await asyncio.create_subprocess_exec(
            cmd, *args, stdout=asyncio.subprocess.PIPE)
    except Exception as e:
        proc_data_future.set_exception(e)
        raise
    proc_data_future.set_result({'proc': proc, 'pid': proc.pid})
    await read_stream(proc, proc.stdout, stdout_callback)
    return proc

@contextlib.contextmanager
def open_subprocess(cmd, *args, stdout_callback=print):
    loop = asyncio.new_event_loop()
    # needed to use asyncio.subprocess outside the main thread
    asyncio.get_child_watcher().attach_loop(loop)
    threading.Thread(target=loop.run_forever).start()
    proc_data_future = concurrent.futures.Future()
    loop.call_soon_threadsafe(
        loop.create_task,
        stream_subprocess(cmd, *args,
                          proc_data_future=proc_data_future,
                          stdout_callback=stdout_callback))
    proc_data = proc_data_future.result()
    yield proc_data
    async def terminate(proc):
        if proc.returncode is None:
            proc.terminate()
            await proc.wait()
    asyncio.run_coroutine_threadsafe(terminate(proc_data['proc']), loop).result()
    proc_data['returncode'] = proc_data['proc'].returncode
    loop.call_soon_threadsafe(loop.stop)

if __name__ == '__main__':
    import time

    def stdout_callback(data):
        print('STDOUT:', data)

    with open_subprocess('ping', '-c', '4', 'localhost',
                         stdout_callback=stdout_callback) as proc_data:
        for i in range(2):
            print(f'RUNNING SUBPROCESS {proc_data["pid"]}...')
            time.sleep(1)

    print(f'RETURN CODE: {proc_data["returncode"]}')

Способ работы с @contextlib.asynccontextmanager и подпрограммой Process.wait() (ожидает дочерний процесс для завершения, устанавливает и возвращает атрибут returncode):

import asyncio
import contextlib

async def read_stream(proc, stream, callback):
    while proc.returncode is None:
        data = await stream.readline()
        if not data:
            break
        callback(data.decode().rstrip())


async def stream_subprocess(cmd, *args, stdout_callback=print):
    proc = await asyncio.create_subprocess_exec(cmd, *args,
                                                stdout=asyncio.subprocess.PIPE)
    await read_stream(proc, proc.stdout, stdout_callback)
    return proc


@contextlib.asynccontextmanager
async def open_subprocess(cmd, *args, stdout_callback=print):
    try:
        proc = await stream_subprocess(cmd, *args, stdout_callback=stdout_callback)
        yield proc
    finally:
        await proc.wait()

if __name__ == '__main__':
    import time

    def stdout_callback(data):
        print('STDOUT:', data)


    async def main():
        async with open_subprocess('ping', '-c', '4', 'localhost',
                                   stdout_callback=stdout_callback) as proc:
            # The following code only runs after proc completes
            for i in range(2):
                print(f'RUNNING SUBPROCESS {proc.pid}...')
                time.sleep(1)

        print(f'RETURN CODE: {proc.returncode}')

    asyncio.run(main())

Пример рабочего вывода:

STDOUT: PING localhost (127.0.0.1): 56 data bytes
STDOUT: 64 bytes from 127.0.0.1: icmp_seq=0 ttl=64 time=0.048 ms
STDOUT: 64 bytes from 127.0.0.1: icmp_seq=1 ttl=64 time=0.074 ms
STDOUT: 64 bytes from 127.0.0.1: icmp_seq=2 ttl=64 time=0.061 ms
STDOUT: 64 bytes from 127.0.0.1: icmp_seq=3 ttl=64 time=0.067 ms
STDOUT: 
STDOUT: --- localhost ping statistics ---
STDOUT: 4 packets transmitted, 4 packets received, 0.0% packet loss
STDOUT: round-trip min/avg/max/stddev = 0.048/0.062/0.074/0.010 ms
RUNNING SUBPROCESS 35439...
RUNNING SUBPROCESS 35439...
RETURN CODE: 0

Process finished with exit code 0