Method _bg_refresh

mycli/completion_refresher.py:53–100

(
        self,
        sqlexecute: SQLExecute,
        callbacks: Callable | list[Callable],
        completer_options: dict,
    )

        return bool(self._completer_thread and self._completer_thread.is_alive())

    def _bg_refresh(
        self,
        sqlexecute: SQLExecute,
        callbacks: Callable | list[Callable],
        completer_options: dict,
    ) -> None:
        completer = SQLCompleter(**completer_options)

        # Create a new sqlexecute method to populate the completions.
        e = sqlexecute
        try:
            executor = SQLExecute(
                e.dbname,
                e.user,
                e.password,
                e.host,
                e.port,
                e.socket,
                e.character_set,
                e.local_infile,
                e.ssl,
            )
        except pymysql.err.OperationalError:
            return

        # If callbacks is a single function then push it into a list.
        if callable(callbacks):
            callbacks = [callbacks]

        while 1:
            for refresher in self.refreshers.values():
                refresher(completer, executor)
                if self._restart_refresh.is_set():
                    self._restart_refresh.clear()
                    break
            else:
                # Break out of while loop if the for loop finishes natually
                # without hitting the break statement.
                break

            # Start over the refresh from the beginning if the for loop hit the
            # break statement.
            continue

        for callback in callbacks:
            callback(completer)

        executor.close()


def refresher(name: str, refreshers: dict = CompletionRefresher.refreshers) -> Callable: