Python async input

ainput.py

import asyncio
from concurrent.futures import ThreadPoolExecutor


async def ainput(prompt: str = "") -> str:
    with ThreadPoolExecutor(1, "AsyncInput") as executor:
        return await asyncio.get_event_loop().run_in_executor(executor, input, prompt)


async def main():
    name = await ainput("What's your name? ")
    print(f"Hello, {name}!")


asyncio.run(main())

Cannot remember why, as years have passed since.

This snippet is being useful in Pyrogram, in which I indeed eventually switched to using input and getpass as functions to be run in the executor. The use of sys.stdin.readline and the thread initializer definitely looks unnecessary.

Gist updated with a simplified version.

asyncio.to_thread was introduced with Python 3.9.

This makes it much easier to call blocking functions from async code.

async def ainput(prompt: str = ""):
    return await asyncio.to_thread(input, prompt)

very nice idea for asynchrinous inputs, Thanks!