Merge pull request #1170 from lazmond3/feature/add-fzf-history

Add fzf history search feature

changelog.md

@@ -11,6 +11,8 @@ Internal:
 Features:
 ---------
 
+* Added fzf history search functionality. The feature can switch between the old implementation and the new one based on the presence of the fzf binary.
+
 
 1.27.2 (2024/04/03)
 ===================
@@ -24,7 +26,6 @@ Bug Fixes:
 1.27.1 (2024/03/28)
 ===================
 
-
 Bug Fixes:
 ----------
 

mycli/AUTHORS

@@ -97,6 +97,7 @@ Contributors:
   * Zhanze Wang
   * Houston Wong
   * Mohamed Rezk
+  * Ryosuke Kazami
 
 
 Created by:

mycli/key_bindings.py

@@ -3,6 +3,8 @@
 from prompt_toolkit.filters import completion_is_selected, emacs_mode
 from prompt_toolkit.key_binding import KeyBindings
 
+from .packages.toolkit.fzf import search_history
+
 _logger = logging.getLogger(__name__)
 
 
@@ -101,6 +103,12 @@ def _(event):
                 cursorpos_abs -= 1
             b.cursor_position = min(cursorpos_abs, len(b.text))
 
+    @kb.add('c-r', filter=emacs_mode)
+    def _(event):
+        """Search history using fzf or default reverse incremental search."""
+        _logger.debug('Detected <C-r> key.')
+        search_history(event)
+
     @kb.add('enter', filter=completion_is_selected)
     def _(event):
         """Makes the enter key work as the tab key only when showing the menu.

mycli/main.py

@@ -36,14 +36,14 @@
 from prompt_toolkit.layout.processors import (HighlightMatchingBracketProcessor,
                                               ConditionalProcessor)
 from prompt_toolkit.lexers import PygmentsLexer
-from prompt_toolkit.history import FileHistory
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 
 from .packages.special.main import NO_QUERY
 from .packages.prompt_utils import confirm, confirm_destructive_query
 from .packages.tabular_output import sql_format
 from .packages import special
 from .packages.special.favoritequeries import FavoriteQueries
+from .packages.toolkit.history import FileHistoryWithTimestamp
 from .sqlcompleter import SQLCompleter
 from .clitoolbar import create_toolbar_tokens_func
 from .clistyle import style_factory, style_factory_output
@@ -626,7 +626,7 @@ def run_cli(self):
         history_file = os.path.expanduser(
             os.environ.get('MYCLI_HISTFILE', '~/.mycli-history'))
         if dir_path_exists(history_file):
-            history = FileHistory(history_file)
+            history = FileHistoryWithTimestamp(history_file)
         else:
             history = None
             self.echo(

mycli/packages/filepaths.py

@@ -100,7 +100,7 @@ def guess_socket_location():
         for r, dirs, files in os.walk(directory, topdown=True):
             for filename in files:
                 name, ext = os.path.splitext(filename)
-                if name.startswith("mysql") and ext in ('.socket', '.sock'):
+                if name.startswith("mysql") and name != "mysqlx" and ext in ('.socket', '.sock'):
                     return os.path.join(r, filename)
             dirs[:] = [d for d in dirs if d.startswith("mysql")]
     return None

mycli/packages/toolkit/fzf.py

@@ -0,0 +1,45 @@
+from shutil import which
+
+from pyfzf import FzfPrompt
+from prompt_toolkit import search
+from prompt_toolkit.key_binding.key_processor import KeyPressEvent
+
+from .history import FileHistoryWithTimestamp
+
+
+class Fzf(FzfPrompt):
+    def __init__(self):
+        self.executable = which("fzf")
+        if self.executable:
+            super().__init__()
+
+    def is_available(self) -> bool:
+        return self.executable is not None
+
+
+def search_history(event: KeyPressEvent):
+    buffer = event.current_buffer
+    history = buffer.history
+
+    fzf = Fzf()
+
+    if fzf.is_available() and isinstance(history, FileHistoryWithTimestamp):
+        history_items_with_timestamp = history.load_history_with_timestamp()
+
+        formatted_history_items = []
+        original_history_items = []
+        for item, timestamp in history_items_with_timestamp:
+            formatted_item = item.replace('\n', ' ')
+            timestamp = timestamp.split(".")[0] if "." in timestamp else timestamp
+            formatted_history_items.append(f"{timestamp}  {formatted_item}")
+            original_history_items.append(item)
+
+        result = fzf.prompt(formatted_history_items, fzf_options="--tiebreak=index")
+
+        if result:
+            selected_index = formatted_history_items.index(result[0])
+            buffer.text = original_history_items[selected_index]
+            buffer.cursor_position = len(buffer.text)
+    else:
+        # Fallback to default reverse incremental search
+        search.start_search(direction=search.SearchDirection.BACKWARD)

mycli/packages/toolkit/history.py

@@ -0,0 +1,52 @@
+import os
+from typing import Iterable, Union, List, Tuple
+
+from prompt_toolkit.history import FileHistory
+
+_StrOrBytesPath = Union[str, bytes, os.PathLike]
+
+
+class FileHistoryWithTimestamp(FileHistory):
+    """
+    :class:`.FileHistory` class that stores all strings in a file with timestamp.
+    """
+
+    def __init__(self, filename: _StrOrBytesPath) -> None:
+        self.filename = filename
+        super().__init__(filename)
+
+    def load_history_with_timestamp(self) -> List[Tuple[str, str]]:
+        """
+        Load history entries along with their timestamps.
+
+        Returns:
+            List[Tuple[str, str]]: A list of tuples where each tuple contains
+                                   a history entry and its corresponding timestamp.
+        """
+        history_with_timestamp: List[Tuple[str, str]] = []
+        lines: List[str] = []
+        timestamp: str = ""
+
+        def add() -> None:
+            if lines:
+                # Join and drop trailing newline.
+                string = "".join(lines)[:-1]
+                history_with_timestamp.append((string, timestamp))
+
+        if os.path.exists(self.filename):
+            with open(self.filename, "rb") as f:
+                for line_bytes in f:
+                    line = line_bytes.decode("utf-8", errors="replace")
+
+                    if line.startswith("#"):
+                        # Extract timestamp
+                        timestamp = line[2:].strip()
+                    elif line.startswith("+"):
+                        lines.append(line[1:])
+                    else:
+                        add()
+                        lines = []
+
+                add()
+
+        return list(reversed(history_with_timestamp))

requirements-dev.txt

@@ -14,4 +14,4 @@ pyperclip>=1.8.1
 importlib_resources>=5.0.0
 pyaes>=1.6.1
 sqlglot>=5.1.3
-setuptools
+setuptools<=71.1.0

setup.py

@@ -29,7 +29,8 @@
     'configobj >= 5.0.5',
     'cli_helpers[styles] >= 2.2.1',
     'pyperclip >= 1.8.1',
-    'pyaes >= 1.6.1'
+    'pyaes >= 1.6.1',
+    'pyfzf >= 0.3.1',
 ]
 
 if sys.version_info.minor < 9: