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
python类CommandLineInterface()的实例源码
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