I have an app, which has to make a get request to my API every 4 seconds to stay authorized. My issue is, it uses a whole lot of input() which is blocking the thread to make the request. I tried to write a whole module to combat this, but it is still blocking. I have spent so long trying to get a non-blocking input. I have tried almost everything on SO. Here is the class I wrote for non-blocking input. (it has a function nbput() that functions almost entirely like python's input())

from pynput import keyboard
from pynput.keyboard import Key
import threading
import sys
from functools import partial

uinput = ''
lastKey = ''
modifierKey = ''


class NonBlockingInput:
    arg0 = ''

    def __init__(self):
        global listener
        listener = keyboard.Listener(on_press=self.on_press, on_release=self.on_release)
        print('1')

    def on_press(self, key):
        global lastKey
        global modifierKey
        try:
            sys.stdout.write(key.char)
            lastKey = key.char
        except AttributeError:
            if key == Key.space:
                lastKey = ' '
                sys.stdout.write(' ')
                return
            modifierKey = key

    def on_release(self, key):
        pass

    def nbinput(self, prompt):
        global uinput
        global listener
        global lastKey
        global modifierKey
        global arg0
        listener.start()
        sys.stdout.write(prompt)
        while True:
            if modifierKey == Key.enter:
                sys.stdout.write('\n')
                value_returned = partial(self.retrieved_data_func, arg0)
                break
            elif modifierKey == Key.backspace:
                spaceString = ''
                for _ in range(0, len(uinput)):
                    spaceString  = ' '
                uinput = uinput[:-1]
                sys.stdout.write('\r')
                sys.stdout.write(spaceString)
                sys.stdout.write('\r')
                sys.stdout.write(uinput)
                modifierKey = ''
            else:
                uinput  = lastKey
                lastKey = ''

    def retrieved_data_func(self):
        arg0 = 0
        return arg0


def nbput(prompt=''):
    global collectionThread
    nonBlockingInput = NonBlockingInput()
    collectionThread = threading.Thread(nonBlockingInput.nbinput(prompt))
    collectionThread.start()
    return NonBlockingInput.retrieved_data_func()


if __name__ == '__main__':
    print(nbput())```

CodePudding user response：

I found a solution: I create the authThread, and then call main.

CodePudding user response：

All that plumbing seems to get in the way of something that's not that complicated:

from time import sleep
from pynput import keyboard


def key_press(key):
    print(f'Pressed {key}')


def api_call():
    print('Making API call...')


listener = keyboard.Listener(on_press=key_press)
listener.start()


while True:
    api_call()
    sleep(4)

This script "calls an API" once every 4 seconds, while handling user input as it is coming in. It uses pyinput as you are, so it should work on whatever platform you're on.

Since the 'API call' here is just a print statement, nothing fancy is needed, but you could of course make the call asynchronously as well. The sleep(4) is just there to simulate 4 seconds of otherwise blocking activity.

You responded that you wanted it all in a class, which makes some sense in avoiding globals, but it's not a class you'd want to instance more than once (or even once).

Here's an example that seems to do what you want while avoiding some of the complexity:

import sys
from time import sleep
from pynput import keyboard
import winsound
try:
    import msvcrt
except ImportError:
    msvcrt = None
    import termios


class NBI:
    text = ''
    listener = None

    @classmethod
    def start(cls):
        if cls.listener is None:
            cls.listener = keyboard.Listener(on_press=cls.key_press)
            cls.listener.start()
        else:
            raise Exception('Cannot start NBI twice.')

    @staticmethod
    def flush_input():
        if msvcrt is not None:
            while msvcrt.kbhit():
                msvcrt.getch()
        else:
            termios.tcflush(sys.stdin, termios.TCIOFLUSH)

    @classmethod
    def key_press(cls, key):
        if key == keyboard.Key.enter:
            sys.stdout.write('\n')
            cls.flush_input()
            cls.listener.stop()
            cls.listener = None
        if key == keyboard.Key.backspace:
            cls.text = cls.text[:-1]
            sys.stdout.write('\b \b')
            sys.stdout.flush()
        elif hasattr(key, 'char'):
            cls.text  = key.char
            sys.stdout.write(key.char)
            sys.stdout.flush()
            cls.flush_input()


def api_call():
    winsound.Beep(440, 200)  # "API call"


NBI.start()
while NBI.listener:
    api_call()
    for _ in range(4):
        sleep(1)
        if not NBI.listener:
            break


print(f'You entered: {NBI.text}')