Answer python input from within on another thread

Question

I want to answer a input() from another thread of the same process on python from within the code.

This is the code:

import sys
import threading

def threaded(fn):
    def wrapper(*args, **kwargs):
        thread = threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=True)
        thread.start()
        return thread
    return wrapper

@threaded
def answer():
    time.sleep(2)
    sys.stdin.write('to be inputed')


answer()
x = input('insert a value: ')
print(f'value inserted: {x}')  # excpeted print: 'value inserted: to be inputed'

But I think its not possbile because I receive this error:

Exception in thread Thread-1:
Traceback (most recent call last):
  File "teste.py", line 80, in answer
sys.stdin.write('to be inputed')
io.UnsupportedOperation: not writable

It's hard to explain why I want that, but sometimes the user will input the value and sometimes it will come from another input source (telegram). So this second thread should be able to input the value and release the code execution.

I also can't change the input() part of the code because its from inside a library, so it need to be this way: input('insert a value: ')

Is there a way to achive that?

Answer 1

The simple answer is that if you replace sys.stdin with your own variable, then input uses that instead.

However, then you've lost your original stdin, so you need start a new process to listen for user input, since you said:

but sometimes the user will input the value

This needs to be another process rather than a thread since it needs to be killed when you want to restore the original stdin, and killing the process interrupts it mid-readline.

Here is a working version of the code with the mock object implemented. The region inside the with block is where stdin has been replaced.

import sys
import time
import multiprocessing
import threading

class MockStdin:
    def __init__(self):
        self.queue = None
        self.real_stdin = sys.stdin
        self.relay_process = None

    def readline(self):
        # when input() is called, it calls this function
        return self.queue.get()

    def writeline(self, s):
        # for input from elsewhere in the program
        self.queue.put(s)

    def relay_stdin(self):
        # for input from the user
        my_stdin = open(0)  # this is a new process so it needs its own stdin

        try:
            while True:
                inp = my_stdin.readline()
                self.queue.put(inp)
        except KeyboardInterrupt:
            # when killed, exit silently
            pass

    def __enter__(self):
        # when entering the `with` block, start replace stdin with self and relay real stdin
        self.queue = multiprocessing.Queue()

        self.relay_process = multiprocessing.Process(target=self.relay_stdin)
        self.relay_process.start()
        sys.stdin = self

    def __exit__(self, exc_type=None, exc_val=None, exc_tb=None):
        # when exiting the `with` block, put stdin back and stop relaying
        sys.stdin = self.real_stdin

        self.relay_process.terminate()
        self.relay_process.join()
        
    def __getstate__(self):
        # this is needed for Windows - credit to Leonardo Rick for this fix
        self_dict = self.__dict__.copy()
        del self_dict['real_stdin']
        return self_dict



def threaded(fn):
    def wrapper(*args, **kwargs):
        thread = threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=True)
        thread.start()
        return thread

    return wrapper

if __name__ == '__main__':
    mock = MockStdin()


    @threaded
    def answer():
        time.sleep(2)
        # use mock to write to stdin
        mock.writeline('to be inputed')


    answer()

    with mock:
        # inside `with` block, stdin is replaced
        x = input('insert a value: ')
        print(f'\nvalue inserted: {x}')

    answer()

    # __enter__ and __exit__ can also be used
    mock.__enter__()
    x = input('insert a value: ')
    print(f'\nvalue inserted: {x}')
    mock.__exit__()

    # now outside the `with` block, stdin is back to normal
    x = input('insert another (stdin should be back to normal now): ')
    print(f'value inserted: {x}')