Python prompt_toolkit 模块，CommandLineInterface() 实例源码

def _run(self):
        args = {
            'history': history.InMemoryHistory(),
            'enable_history_search': True,
            'enable_open_in_editor': True,
            'lexer': lexer,
            'completer': get_completer(self._environment, self._couch_server),
            'style': style,
        }
        while True:
            try:
                cli = pt.CommandLineInterface(application=shortcuts.create_prompt_application(self.prompt, **args),
                                              eventloop=shortcuts.create_eventloop())
                self._environment.cli = cli
                cmd_text = cli.run().text.rstrip()
                eval_(self._environment, self._couch_server, cmd_text)
                cli.reset()
            except RuntimeError as e:
                self._environment.output(str(e))
            except (EOFError, KeyboardInterrupt):
                self._environment.output('Exiting...')
                break

def __init__(self, parser, engine, options=None):
        self.parser = parser
        self.engine = engine
        self.options = options if options is not None else {}
        util.ensure_data_dir_exists()
        application = create_prompt_application(
            message='> ',
            lexer=PygmentsLexer(SqlLexer),
            history=FileHistory(os.path.expanduser('~/.aq/history')),
            completer=AqCompleter(schemas=engine.available_schemas, tables=engine.available_tables),
            auto_suggest=AutoSuggestFromHistory(),
            validator=QueryValidator(parser),
            on_abort=AbortAction.RETRY,
        )
        loop = create_eventloop()
        self.cli = CommandLineInterface(application=application, eventloop=loop)
        self.patch_context = self.cli.patch_stdout_context()

def run():
    validate_osenvironment()
    print_banner()

    cli_buffer = OSBuffer()
    ltobj = OSLayout(multiwindow=False)

    application = Application(
        style=PygmentsStyle(OSStyle),
        layout=ltobj.layout,
        buffers=cli_buffer.buffers,
        on_exit=AbortAction.RAISE_EXCEPTION,
        key_bindings_registry=OSKeyBinder.registry)


    cli = CommandLineInterface(application=application,
                               eventloop=create_eventloop())

    while True:
        try:
            document = cli.run(reset_current_buffer=True)
            process_document(document)
        except KeyboardInterrupt:
            # A keyboardInterrupt generated possibly due to Ctrl-C
            print "Keyboard Interrupt Generated"
            continue
        except EOFError:
            print "cntl-D"
            sys.exit()

def _get_cli(self, loop):
        global history
        return CommandLineInterface(
            application=create_prompt_application(
                multiline=Condition(lambda: self._multi_line),
                get_prompt_tokens=self._get_prompt,
                history=history,
                wrap_lines=True,
            ),
            eventloop=loop,
        )

def run(self):
        style = style_from_dict({
            Token.Prompt: 'bold',
            Token.Toolbar: '#ccc bg:#333',
            Token.Name: '#fff bold bg:#333',
        })

        history = InMemoryHistory()
        eventloop = create_eventloop()
        app = create_prompt_application(history=history, style=style,
                                        get_bottom_toolbar_tokens=self.get_bottom_toolbar_tokens,
                                        get_prompt_tokens=self.get_prompt_tokens)
        self.cli = CommandLineInterface(app, eventloop)

        with self.cli.patch_stdout_context(raw=True):
            while True:
                try:
                    self.cli.run()
                    doc = self.cli.return_value()
                    if doc is None:
                        return
                    cmd = shlex.split(doc.text)
                    app.buffer.reset(append_to_history=True)

                    if not cmd:
                        continue
                    elif cmd[0] in ('exit', 'quit'):
                        self.q.put(Exit())
                        return
                    elif cmd[0] == 'help':
                        print('Help text forthcoming.')
                    elif cmd[0] == 'skip':
                        self.q.put(Skip())
                    elif cmd[0] == 'set':
                        self.q.put(Set(cmd[1], ast.literal_eval(' '.join(cmd[2:]))))
                    else:
                        print('Unknown command. Try \'help\'.')
                except KeyboardInterrupt:
                    continue
                except EOFError:
                    self.q.put(Exit())
                    return
                except Exception as err:
                    print(err)
                    self.q.put(Exit())
                    return