Python prompt_toolkit.history 模块，FileHistory() 实例源码

def main():
    """ 
         Start the Slack Client 
    """
    os.system("clear; figlet 'Slack Gitsin' | lolcat")
    history = FileHistory(os.path.expanduser("~/.slackHistory"))
    while True:
        text = prompt("slack> ", history=history,
                      auto_suggest=AutoSuggestFromHistory(),
                      on_abort=AbortAction.RETRY,
                      style=DocumentStyle,
                      completer=Completer(fuzzy_match=False,
                                          text_utils=TextUtils()),
                      complete_while_typing=Always(),
                      get_bottom_toolbar_tokens=get_bottom_toolbar_tokens,
                      key_bindings_registry=manager.registry,
                      accept_action=AcceptAction.RETURN_DOCUMENT
        )
        slack = Slack(text)
        slack.run_command()

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 tagger_repl(tagger, **kwargs):
    mxlen = int(kwargs.get('mxlen', 100))
    maxw = int(kwargs.get('maxw', 100))
    #zeropad = int(kwargs.get('zeropad', 0))
    prompt_name = kwargs.get('prompt', 'class> ')
    history_file = kwargs.get('history_file', '.history')
    history = FileHistory(history_file)

    while True:
        text = prompt(prompt_name, history=history)
        text = text.strip()
        if text == 'quit':
            break
        try:
            tokens = text.split(' ')
            best = tagger.predict_text(tokens, mxlen=mxlen, maxw=maxw)
            print(best)
        except Exception as e:
            logging.exception('Error')

def test(self):
        self._load_auth()
        history = FileHistory('.history')

        project_id = self._get_current_project_id()
        bot = self._load_bot()

        while True:
            try:
                line = prompt('BotHub> ', history=history)
                if not line:
                    continue
                event = make_event(line)
                context = {}
                bot.handle_message(event, context)
            except (EOFError, KeyboardInterrupt):
                break
            except Exception:
                traceback.print_exc()

def __init__(self, client, multiline, metadata, *args, **kwargs):
        @Condition
        def is_multiline():
            if not multiline:
                return False

            text = self.document.text
            return not query_is_finished(text, multiline)

        super(CLIBuffer, self).__init__(
            *args,
            completer=CHCompleter(client, metadata),
            history=FileHistory(
                filename=os.path.expanduser('~/.clickhouse-cli_history')
            ),
            enable_history_search=True,
            accept_action=AcceptAction.RETURN_DOCUMENT,
            is_multiline=is_multiline,
            **kwargs
        )

def main(args):
    os.environ[ENV_ADDITIONAL_USER_AGENT] = 'AZURECLISHELL/' + __version__

    parser = argparse.ArgumentParser(prog='az-shell')
    parser.add_argument(
        '--style', dest='style', help='the colors of the shell',
        choices=get_options())
    args = parser.parse_args(args)

    azure_folder = cli_config_dir()
    if not os.path.exists(azure_folder):
        os.makedirs(azure_folder)

    ACCOUNT.load(os.path.join(azure_folder, 'azureProfile.json'))
    CONFIG.load(os.path.join(azure_folder, 'az.json'))
    SESSION.load(os.path.join(azure_folder, 'az.sess'), max_age=3600)

    config = SHELL_CONFIGURATION
    shell_config_dir = azclishell.configuration.get_config_dir

    if args.style:
        given_style = args.style
        config.set_style(given_style)
    else:
        given_style = config.get_style()

    style = style_factory(given_style)

    if config.BOOLEAN_STATES[config.config.get('DEFAULT', 'firsttime')]:
        print("When in doubt, ask for 'help'")
        config.firsttime()

    shell_app = Shell(
        completer=AZCOMPLETER,
        lexer=AzLexer,
        history=FileHistory(
            os.path.join(shell_config_dir(), config.get_history())),
        app=APPLICATION,
        styles=style
    )
    shell_app.run()

def __init__(self, refresh_resources=True):
        shell_dir = os.path.expanduser("~/.kube/shell/")
        self.history = FileHistory(os.path.join(shell_dir, "history"))
        if not os.path.exists(shell_dir):
            os.makedirs(shell_dir)
        self.toolbar = Toolbar(self.get_cluster_name, self.get_namespace, self.get_user, self.get_inline_help)

def __init__(self, config, completer):
        self._cli = None
        self._env = os.environ.copy()
        self.history = InMemoryHistory()
        self.file_history = FileHistory(
                "{}/history".format(AzureShellCache.Instance().get('base_dir')))
        self._config = config
        self.completer = completer

def classifier_repl(classifier, **kwargs):

    mxlen = int(kwargs.get('mxlen', 100))
    k = int(kwargs.get('k', 1))
    thresh = float(kwargs.get('thresh', 0.0))
    zeropad = int(kwargs.get('zeropad', 0))
    prompt_name = kwargs.get('prompt', 'class> ')
    history_file = kwargs.get('history_file', '.history')
    history = FileHistory(history_file)

    while True:
        text = prompt(prompt_name, history=history)
        text = text.strip()
        if text == 'quit':
            break
        try:
            tokens = text.split(' ')
            outcomes = classifier.classify_text(tokens, mxlen=mxlen, zeropad=zeropad)
            k = min(k, len(outcomes))
            probs = outcomes[:k]
            for prob_i in probs:
                if prob_i[1] > thresh:
                    print('Guess [%s]: %.3f' % prob_i)

        except Exception as e:
            logging.exception('Error')

def read_stdin():
    while True:
        try:
            if get_variable('HISTORY'):
                fileHistory = FileHistory(get_variable('HISTORY'))
            else:
                fileHistory = None

            line = prompt(program + " > ", history=fileHistory, completer=CommandCompleter())
            process_line(line)
        except EOFError:
            break

def cmdloop(self):

        history_location = os.path.join(os.path.dirname(os.path.abspath(__file__)),".history")
        history = FileHistory(history_location)

        print(frontend_message.get_welcome_message())
        while True:
            try:
                while True:
                    text = prompt(self.prompt, lexer=PythonLexer,
                                  completer=self.completer,
                                  style=DocumentStyle, history=history)
                    res = self.processor.forward(text)

                    if(isinstance(res, str)):
                        res = res.strip()
                        if(len(res)>0):
                            print(res)
                    elif res is not None:
                        print(res)


            except KeyboardInterrupt as ex:
                print("^C")
            except EOFError:
                print(frontend_message.get_bye_message())
                break
            except Exception as ex:
                print(ex)

def _setup_cli_history(self):
        self.cli_history = FileHistory('.cli_history')

def shell():
    c = Chitin()

    cmd_history = FileHistory(os.path.expanduser('~') + '/.chitin.history')
    print(WELCOME)
    message = VERSION

    def get_bottom_toolbar_tokens(cli):
        return [(Token.Toolbar, ' '+message)]

    style = style_from_dict({
        Token.Toolbar: '#ffffff bg:#333333',
    })
    completer = SystemCompleter()
    del completer.completers["executable"]

    # Check whether files in and around the current directory have been changed...
    for failed in util.check_integrity_set(set(".")):
        print("[WARN] '%s' has been modified outside of lab book." % failed)
    try:
        while True:
            cmd_str = ""
            while len(cmd_str.strip()) == 0:
                cmd_str = prompt(u'===> ',
                        history=cmd_history,
                        auto_suggest=AutoSuggestFromHistory(),
                        completer=completer,
                        lexer=PygmentsLexer(BashLexer),
                        get_bottom_toolbar_tokens=get_bottom_toolbar_tokens,
                        style=style,
                        on_abort=AbortAction.RETRY,
                )


            fields = cmd_str.split(" ")
            command_set = [" ".join(fields)]

            skip, special_command_set = c.attempt_special(cmd_str)
            if skip:
                continue
            if len(special_command_set) > 0:
                command_set = special_command_set

            #####################################
            handled = c.super_handle(command_set)
            if handled:
                if "message" in handled:
                    message = handled["message"]
                else:
                    message = VERSION
            #####################################

    except EOFError:
        print("Bye!")

def __init__(self, uri, auth, secure=True, verbose=False):
        try:
            self.driver = GraphDatabase.driver(uri, auth=auth, encrypted=secure)
        except ServiceUnavailable as error:
            raise ConsoleError("Could not connect to {} ({})".format(uri, error))
        self.uri = uri
        self.history = FileHistory(HISTORY_FILE)
        self.prompt_args = {
            "history": self.history,
            "lexer": PygmentsLexer(CypherLexer),
            "style": style_from_pygments(VimStyle, {
                Token.Prompt: "#ansi{}".format(self.prompt_colour.replace("cyan", "teal")),
                Token.TxCounter: "#ansi{} bold".format(self.tx_colour.replace("cyan", "teal")),
            })
        }
        self.lexer = CypherLexer()
        self.result_writer = TabularResultWriter()
        if verbose:
            from .watcher import watch
            self.watcher = watch("neo4j.bolt")

        self.commands = {

            "//": self.set_multi_line,
            "/e": self.edit,

            "/?": self.help,
            "/h": self.help,
            "/help": self.help,

            "/x": self.exit,
            "/exit": self.exit,

            "/r": self.run_read_tx,
            "/read": self.run_read_tx,
            "/w": self.run_write_tx,
            "/write": self.run_write_tx,

            "/csv": self.set_csv_result_writer,
            "/table": self.set_tabular_result_writer,
            "/tsv": self.set_tsv_result_writer,

            "/config": self.config,
            "/kernel": self.kernel,

        }
        self.session = None
        self.tx = None
        self.tx_counter = 0

def main(style=None):
    if APPLICATION.session["az_interactive_active"]:
        logger.warning("You're in the interactive shell already.\n")
        return

    os.environ[ENV_ADDITIONAL_USER_AGENT] = 'AZURECLISHELL/' + __version__

    azure_folder = cli_config_dir()
    if not os.path.exists(azure_folder):
        os.makedirs(azure_folder)

    ACCOUNT.load(os.path.join(azure_folder, 'azureProfile.json'))
    CONFIG.load(os.path.join(azure_folder, 'az.json'))
    SESSION.load(os.path.join(azure_folder, 'az.sess'), max_age=3600)

    config = azclishell.configuration.CONFIGURATION
    shell_config_dir = azclishell.configuration.get_config_dir

    try:
        commands = GatherCommands()
        az_completer = AzCompleter(commands)
    except IOError:  # if there is no cache
        az_completer = None

    if style:
        given_style = style
        config.set_style(given_style)
    else:
        given_style = config.get_style()

    style_obj = style_factory(given_style)

    if config.BOOLEAN_STATES[config.config.get('DEFAULT', 'firsttime')]:
        config.firsttime()

    ask_feedback = False
    if not config.has_feedback() and frequent_user:
        print("\n\nAny comments or concerns? You can use the \'feedback\' command!" +
              " We would greatly appreciate it.\n")
        ask_feedback = True

    shell_app = Shell(
        completer=az_completer,
        lexer=AzLexer,
        history=FileHistory(
            os.path.join(shell_config_dir(), config.get_history())),
        app=APPLICATION,
        styles=style_obj,
        user_feedback=ask_feedback
    )
    shell_app.app.session["az_interactive_active"] = True
    shell_app.run()
    shell_app.app.session["az_interactive_active"] = False

def start_repl(self, quiet: bool) -> None:
        """
            Start the objection repl.
        """

        banner = ("""
     _     _         _   _
 ___| |_  |_|___ ___| |_|_|___ ___
| . | . | | | -_|  _|  _| | . |   |
|___|___|_| |___|___|_| |_|___|_|_|
        |___|(object)inject(ion) v{0}

     Runtime Mobile Exploration
        by: @leonjza from @sensepost
""").format(__version__)

        if not quiet:
            click.secho(banner, bold=True)
            click.secho('[tab] for command suggestions', fg='white', dim=True)

        # the main application loop is here, reading inputs provided by
        # prompt_toolkit and sending it off the the needed handlers
        while True:

            try:

                document = prompt(
                    get_prompt_tokens=self.get_prompt_tokens,
                    completer=self.completer,
                    style=PromptStyle().get_style(),
                    history=FileHistory(os.path.expanduser('~/.objection/objection_history')),
                    auto_suggest=AutoSuggestFromHistory(),
                    on_abort=AbortAction.RETRY,
                    reserve_space_for_menu=4
                )

                # check if this is an exit command
                if document.strip() in ('quit', 'exit', 'bye'):
                    click.secho('Exiting...', dim=True)
                    break

                # if we got the reconnect command, handle just that
                if self.handle_reconnect(document):
                    continue

                # dispatch to the command handler. if something goes horribly
                # wrong, catch it instead of crashing the REPL
                try:

                    # find something to run
                    self.run_command(document)

                except Exception as e:
                    click.secho(('\n\nAn exception occurred while processing the command. If this '
                                 'looks like a code related error, please file a bug report!'), fg='red')
                    click.secho('Error: {0}'.format(e), fg='red', bold=True)

            except (KeyboardInterrupt, EOFError):
                click.secho('Exiting...', dim=True)
                break