How to read from subprocess in aiohttp?

Question

I have a very simple server which reads from a subprocess and passes the data to any opened websockets. The problem I'm running into is that my approach to reading from the subprocess seems to break aiohttp in a way that I can't seem to follow:

#!/usr/bin/env python3

import asyncio
from aiohttp import web
import subprocess

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    request.app['websockets'].append(ws)
    try:
        async for msg in ws:
            print(msg)
            await asyncio.sleep(1)
    finally:
        request.app['websockets'].remove(ws)
    return ws


async def on_shutdown(app):
    for ws in app['websockets']:
        await ws.close(code=999, message='Server shutdown')

Here's where things go wrong:

async def listen_to_process(app):
    print("listen to process")
    while True:
        print(" looping? ")
        await asyncio.sleep(0.1)
        # the problem seems to be here
        line = await app['process'].stdout.readline()
        # if line:
        #     buffer.append(line)

async def start_background_tasks(app):
    app['process_listener'] = app.loop.create_task(listen_to_process(app))


async def cleanup_background_tasks(app):
    print('cleanup background tasks...')
    app['process_listener'].cancel()
    await app['process_listener']


def init():
    app = web.Application()
    app['websockets'] = []
    app.router.add_get('/updates', websocket_handler)
    cmd = "very long running subprocess"
    app['process'] = subprocess.Popen(cmd.split(" "), stdout=subprocess.PIPE)
    app.on_startup.append(start_background_tasks)
    app.on_cleanup.append(cleanup_background_tasks)
    app.on_shutdown.append(on_shutdown)
    return app


web.run_app(init())

So, my question is: how can I read lines from stdout in a loop in the background of my application? Thank you kindly for any pointers.

Answer 1

Please use asyncio subprocess API:

app['process'] = await asyncio.create_subprocess_exec(
                           shlex.split(cmd), stdout=subprocess.PIPE)

Small neat: use shlex module to safely split a command line into a list of arguments.