Set minimum times for transient toolbar messages

changelog.md

@@ -1,3 +1,11 @@
+Upcoming (TBD)
+==============
+
+Features
+---------
+* Set minimum times for transient toolbar messages.
+
+
 2.2.0 (2026/07/11)
 ==============
 

mycli/client.py

@@ -155,7 +155,7 @@ def __init__(
                 self.echo("Error: Unable to open the audit log file. Your queries will not be logged.", err=True, fg="red")
                 self.logfile = False
 
-        self.completion_refresher = CompletionRefresher()
+        self.completion_refresher = CompletionRefresher(self._invalidate_prompt_session)
         self.prefetch_schemas_mode = c["main"].get("prefetch_schemas_mode", "always") or "always"
         raw_prefetch_list = c["main"].as_list("prefetch_schemas_list") if "prefetch_schemas_list" in c["main"] else []
         self.prefetch_schemas_list = [s.strip() for s in raw_prefetch_list if s and s.strip()]
@@ -201,6 +201,10 @@ def __init__(
         self.destructive_keywords = destructive_keywords_from_config(c)
         special.set_destructive_keywords(self.destructive_keywords)
 
+    def _invalidate_prompt_session(self) -> None:
+        if self.prompt_session:
+            self.prompt_session.app.invalidate()
+
     def close(self) -> None:
         try:
             self.schema_prefetcher.stop()

mycli/completion_refresher.py

@@ -1,4 +1,5 @@
 import threading
+from time import monotonic
 from typing import Callable
 
 import pymysql
@@ -8,13 +9,18 @@
 from mycli.sqlcompleter import SQLCompleter
 from mycli.sqlexecute import ServerSpecies, SQLExecute
 
+MIN_COMPLETION_REFRESH_MESSAGE_SECONDS = 1.0
+
 
 class CompletionRefresher:
     refreshers: dict = {}
 
-    def __init__(self) -> None:
+    def __init__(self, invalidate_app: Callable[[], None] | None = None) -> None:
         self._completer_thread: threading.Thread | None = None
         self._restart_refresh = threading.Event()
+        self._refresh_visible_until = 0.0
+        self._visibility_timer: threading.Timer | None = None
+        self._invalidate_app = invalidate_app
 
     def refresh(
         self,
@@ -36,10 +42,14 @@ def refresh(
         if completer_options is None:
             completer_options = {}
 
-        if self.is_refreshing():
+        if self._thread_is_alive():
             self._restart_refresh.set()
             return [SQLResult(status="Auto-completion refresh restarted.")]
         else:
+            if self._visibility_timer is not None:
+                self._visibility_timer.cancel()
+                self._visibility_timer = None
+            self._refresh_visible_until = monotonic() + MIN_COMPLETION_REFRESH_MESSAGE_SECONDS
             self._completer_thread = threading.Thread(
                 target=self._bg_refresh, args=(executor, callbacks, completer_options), name="completion_refresh"
             )
@@ -48,6 +58,9 @@ def refresh(
             return [SQLResult(status="Auto-completion refresh started in the background.")]
 
     def is_refreshing(self) -> bool:
+        return self._thread_is_alive() or monotonic() < self._refresh_visible_until
+
+    def _thread_is_alive(self) -> bool:
         return bool(self._completer_thread and self._completer_thread.is_alive())
 
     def _bg_refresh(
@@ -73,31 +86,53 @@ def _bg_refresh(
                 e.ssl,
             )
         except pymysql.err.OperationalError:
+            self._finish_refreshing()
             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()
+        try:
+            # 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
-            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
+                # Start over the refresh from the beginning if the for loop hit the
+                # break statement.
+                continue
 
-        for callback in callbacks:
-            callback(completer)
+            for callback in callbacks:
+                callback(completer)
+        finally:
+            executor.close()
+            self._finish_refreshing()
 
-        executor.close()
+    def _finish_refreshing(self) -> None:
+        self._invalidate()
+        remaining = self._refresh_visible_until - monotonic()
+        if remaining <= 0:
+            return
+        if self._visibility_timer is not None:
+            self._visibility_timer.cancel()
+        self._visibility_timer = threading.Timer(remaining, self._invalidate_after_visibility_deadline)
+        self._visibility_timer.daemon = True
+        self._visibility_timer.start()
+
+    def _invalidate_after_visibility_deadline(self) -> None:
+        self._visibility_timer = None
+        self._invalidate()
+
+    def _invalidate(self) -> None:
+        if self._invalidate_app is not None:
+            self._invalidate_app()
 
 
 def refresher(name: str, refreshers: dict = CompletionRefresher.refreshers) -> Callable:

mycli/schema_prefetcher.py

@@ -12,6 +12,7 @@
 from enum import Enum
 import logging
 import threading
+from time import monotonic
 from typing import TYPE_CHECKING, Any, Iterable
 
 from mycli.sqlexecute import SQLExecute
@@ -21,6 +22,7 @@
     from mycli.sqlcompleter import SQLCompleter
 
 _logger = logging.getLogger(__name__)
+MIN_PREFETCH_MESSAGE_SECONDS = 1.0
 
 
 class PrefetchMode(str, Enum):
@@ -56,9 +58,11 @@ def __init__(self, mycli: 'MyCli') -> None:
         self._thread: threading.Thread | None = None
         self._cancel = threading.Event()
         self._loaded: set[str] = set()
+        self._prefetch_visible_until = 0.0
+        self._visibility_timer: threading.Timer | None = None
 
     def is_prefetching(self) -> bool:
-        return bool(self._thread and self._thread.is_alive())
+        return bool(self._thread and self._thread.is_alive()) or monotonic() < self._prefetch_visible_until
 
     def clear_loaded(self) -> None:
         """Forget which schemas have been prefetched (used on reset)."""
@@ -69,6 +73,10 @@ def stop(self, timeout: float = 2.0) -> None:
         if self._thread and self._thread.is_alive():
             self._cancel.set()
             self._thread.join(timeout=timeout)
+        self._prefetch_visible_until = 0.0
+        if self._visibility_timer is not None:
+            self._visibility_timer.cancel()
+            self._visibility_timer = None
         self._cancel = threading.Event()
         self._thread = None
 
@@ -105,6 +113,7 @@ def _start(self, schemas: Iterable[str] | None) -> None:
         self.stop()
         queue: list[str] | None = None if schemas is None else list(schemas)
         self._cancel = threading.Event()
+        self._prefetch_visible_until = monotonic() + MIN_PREFETCH_MESSAGE_SECONDS
         self._thread = threading.Thread(
             target=self._run,
             args=(queue,),
@@ -120,7 +129,7 @@ def _run(self, schemas: list[str] | None) -> None:
             executor = self._make_executor()
         except Exception as e:  # pragma: no cover - defensive
             _logger.error('schema prefetch could not open connection: %r', e)
-            self._invalidate_app()
+            self._finish_prefetching()
             return
         try:
             if schemas is None:
@@ -145,7 +154,22 @@ def _run(self, schemas: list[str] | None) -> None:
                 executor.close()
             except Exception:  # pragma: no cover - defensive
                 pass
-            self._invalidate_app()
+            self._finish_prefetching()
+
+    def _finish_prefetching(self) -> None:
+        self._invalidate_app()
+        remaining = self._prefetch_visible_until - monotonic()
+        if remaining <= 0:
+            return
+        if self._visibility_timer is not None:
+            self._visibility_timer.cancel()
+        self._visibility_timer = threading.Timer(remaining, self._invalidate_after_visibility_deadline)
+        self._visibility_timer.daemon = True
+        self._visibility_timer.start()
+
+    def _invalidate_after_visibility_deadline(self) -> None:
+        self._visibility_timer = None
+        self._invalidate_app()
 
     def _prefetch_one(self, executor: SQLExecute, schema: str) -> None:
         _logger.debug('prefetching schema %r', schema)

test/pytests/test_client.py

@@ -158,6 +158,16 @@ def fail() -> None:
     MyCli.close(cli)
 
 
+def test_invalidate_prompt_session_invalidates_prompt_app() -> None:
+    cli = MyCli.__new__(MyCli)
+    invalidate_calls: list[bool] = []
+    cast(Any, cli).prompt_session = SimpleNamespace(app=SimpleNamespace(invalidate=lambda: invalidate_calls.append(True)))
+
+    MyCli._invalidate_prompt_session(cli)
+
+    assert invalidate_calls == [True]
+
+
 def test_run_cli_delegates_to_main_repl(monkeypatch: pytest.MonkeyPatch) -> None:
     cli = MyCli.__new__(MyCli)
     calls: list[MyCli] = []