Improved the REPL mode.

[Champollion9012]

When I press the Arrow keys in REPL mode, the output is garbled. Could you improve that ? like chrome-remote-interface, a javascript project.

[ClericPy]

REPL mode is not an easy task( specifically run await syntax). There are usually different terminals for garbled output.

So it's just a simple test method, I will consider plug-in projects to solve this problem later.

PS: Maybe asyncio.__main__ can give a hand, but it used a different loop.

[ClericPy]

I tried

# copy from import asyncio.__main__
import ast
import asyncio
import code
import concurrent.futures
import inspect
import sys
import types
import typing
import warnings
from asyncio import futures

async def async_repl(repl_locals: dict = None,
                     exit_callback: typing.Callable = None,
                     exit_callback_kwargs: dict = None):
    repl_locals = repl_locals or sys._getframe(1).f_locals

    class AsyncIOInteractiveConsole(code.InteractiveConsole):

        def __init__(self, locals, loop):
            super().__init__(locals)
            self.compile.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT

            self.loop = loop
            self.repl_future_interrupted = False
            self.repl_future = None

        def runsource(self, source, *args, **kwargs):
            if source in {'exit', 'exit()', 'quit', 'quit()'}:
                raise EOFError()
            super().runsource(source, *args, **kwargs)

        def runcode(self, code):
            future = concurrent.futures.Future()

            def callback():

                self.repl_future = None
                self.repl_future_interrupted = False

                func = types.FunctionType(code, self.locals)
                try:
                    coro = func()
                except SystemExit:
                    raise
                except KeyboardInterrupt as ex:
                    self.repl_future_interrupted = True
                    future.set_exception(ex)
                    return
                except BaseException as ex:
                    future.set_exception(ex)
                    return

                if not inspect.iscoroutine(coro):
                    future.set_result(coro)
                    return

                try:
                    self.repl_future = self.loop.create_task(coro)
                    futures._chain_future(self.repl_future, future)
                except BaseException as exc:
                    future.set_exception(exc)

            self.loop.call_soon_threadsafe(callback)

            try:
                return future.result()
            except SystemExit:
                raise
            except BaseException:
                if self.repl_future_interrupted:
                    self.write("\nKeyboardInterrupt\n")
                else:
                    self.showtraceback()

    def run(console):
        try:
            warnings.filterwarnings('ignore',
                                    message=r'^coroutine .* was never awaited$',
                                    category=RuntimeWarning)
            banner = 'Welcome to ichrome asyncio repl.\nNow you can do some actions with `tab` variable.\nInput `quit` or `exit` to shutdown gracefully.\nTry like this:\n\t>>> await tab.goto("http://python.org")\n\t>>> await tab.title\n\t>>> await tab.url\n'
            console.interact(banner=banner, exitmsg='exiting asyncio REPL...')
            if exit_callback:
                exit_callback(**(exit_callback_kwargs or {}))
        except SystemExit:
            pass

    if asyncio.iscoroutinefunction(exit_callback):
        raise TypeError('exit_callback should not be coroutine funtion.')
    try:
        loop = asyncio.get_running_loop()
        for key in {
                '__name__', '__package__', '__loader__', '__spec__',
                '__builtins__', '__file__'
        }:
            repl_locals[key] = globals()[key]
        console = AsyncIOInteractiveConsole(repl_locals, loop)
        await asyncio.get_running_loop().run_in_executor(None, run, console)
    except (KeyboardInterrupt, EOFError):
        if console.repl_future and not console.repl_future.done():
            console.repl_future.cancel()
            console.repl_future_interrupted = True

async def test():
    await async_repl(exit_callback=lambda: print('bye'))

if __name__ == "__main__":
    asyncio.run(test())

But this can not catch theKeyboardInterrupt error raised from event loop. I need to read the code.py again.

import readline will fix garbled arrow, I will fix this