How to pass bytes as non-blocking to Popen in Python

Viewed 46

I came up with the following method to pass bytes to Popen in a non-blocking way.

Is there a better way or a suitable library?

import os
from threading import Thread
from subprocess import Popen
from time import sleep

def get_read_stream(data: bytes):
    r, w = os.pipe()
    def _write():
        with open(w, 'wb') as f:
            f.write(data)
            sleep(5) # <- To check for asynchronicity
        print('closed')
    Thread(target=_write).start()
    return open(r, 'rb')

rs = get_read_stream(b'foobar\n')
proc = Popen(['cat'], stdin=rs)
print('start proc')
proc.wait()
print('end')

P.S.

I've come up with another method. I think this is better because there is less stream to manage.

from threading import Thread
from subprocess import Popen
from time import sleep

def write_to_stream(stream, data: bytes):
    try:
        stream.write(data)
        sleep(5) # <- To check for asynchronicity
    finally:
        stream.close()
        print('closed')

proc = Popen(['cat'], stdin=PIPE)
print('start proc')
Thread(target=write_to_stream, args=(proc.stdin, b'foobar\n')).start()
proc.wait()
print('end')
0 Answers
Related