Add smart SQL completion with history-based frequency tracking

Author: koillinjag-tech

pgcli/completion/__init__.py

@@ -6,5 +6,36 @@
 """
 
 from .history_freq import HistoryFreqTracker, get_history_freq_tracker
+from .smart_completer import SmartPGCompleter
+from ..pgcompleter import PGCompleter
 
-__all__ = ["HistoryFreqTracker", "get_history_freq_tracker"]
+
+def create_completer(smart_completion=True, pgspecial=None, settings=None, smart_completion_history=False):
+    """
+    Factory function to create the appropriate completer based on configuration.
+
+    Args:
+        smart_completion: Base smart completion flag
+        pgspecial: PGSpecial instance
+        settings: Completion settings dict
+        smart_completion_history: Whether to enable history-based smart completion
+
+    Returns:
+        PGCompleter or SmartPGCompleter instance
+    """
+    if smart_completion_history:
+        return SmartPGCompleter(
+            smart_completion=smart_completion,
+            pgspecial=pgspecial,
+            settings=settings,
+            smart_completion_enabled=True
+        )
+    else:
+        return PGCompleter(
+            smart_completion=smart_completion,
+            pgspecial=pgspecial,
+            settings=settings
+        )
+
+
+__all__ = ["HistoryFreqTracker", "get_history_freq_tracker", "SmartPGCompleter", "create_completer"]

pgcli/main.py

@@ -55,8 +55,6 @@
 
 from . import auth
 from .pgcompleter import PGCompleter
-from .completion.smart_completer import SmartPGCompleter
-from .completion.history_freq import get_history_freq_tracker
 from .pgtoolbar import create_toolbar_tokens_func
 from .pgstyle import style_factory, style_factory_output
 from .pgexecute import PGExecute
@@ -270,7 +268,6 @@ def __init__(
 
         # Initialize completer
         smart_completion = c["main"].as_bool("smart_completion")
-        smart_completion_history = c["main"].as_bool("smart_completion_history")
         keyword_casing = c["main"]["keyword_casing"]
         single_connection = single_connection or c["main"].as_bool("always_use_single_connection")
         self.settings = {
@@ -287,18 +284,7 @@ def __init__(
             "alias_map_file": c["main"]["alias_map_file"] or None,
         }
 
-        # Use SmartPGCompleter when history-based completion is enabled
-        if smart_completion_history:
-            completer = SmartPGCompleter(
-                smart_completion,
-                pgspecial=self.pgspecial,
-                settings=self.settings,
-                smart_completion_enabled=True
-            )
-        else:
-            completer = PGCompleter(smart_completion, pgspecial=self.pgspecial, settings=self.settings)
-        self.completer = completer
-        self._smart_completion_history = smart_completion_history
+        self.completer = PGCompleter(smart_completion, pgspecial=self.pgspecial, settings=self.settings)
         self._completer_lock = threading.Lock()
         self.register_special_commands()
 
@@ -428,43 +414,6 @@ def refresh_callback():
             "Toggle verbose errors.",
         )
 
-        self.pgspecial.register(
-            self.toggle_smart_completion,
-            "\\set_smart_completion",
-            "\\set_smart_completion [on|off]",
-            "Toggle history-based smart completion.",
-            arg_type=NO_QUERY,
-            case_sensitive=True,
-        )
-
-    def toggle_smart_completion(self, pattern, **_):
-        """Toggle history-based smart completion."""
-        flag = pattern.strip().lower()
-
-        if flag == "on":
-            self._smart_completion_history = True
-        elif flag == "off":
-            self._smart_completion_history = False
-        else:
-            self._smart_completion_history = not self._smart_completion_history
-
-        # Update the completer
-        if hasattr(self.completer, 'enable_smart_completion'):
-            self.completer.enable_smart_completion(self._smart_completion_history)
-        elif self._smart_completion_history and not hasattr(self.completer, '_history_tracker'):
-            # If enabling but current completer doesn't support it, we need to recreate it
-            from .completion.smart_completer import SmartPGCompleter
-            self.completer = SmartPGCompleter(
-                self.completer.smart_completion,
-                pgspecial=self.pgspecial,
-                settings=self.settings,
-                smart_completion_enabled=True
-            )
-
-        status = "ON" if self._smart_completion_history else "OFF"
-        message = f"History-based smart completion: {status}"
-        return [(None, None, None, message)]
-
     def toggle_verbose_errors(self, pattern, **_):
         flag = pattern.strip()
 
@@ -951,28 +900,6 @@ def execute_command(self, text, handle_closed_connection=True):
                 else:
                     print("Time: %0.03fs" % query.total_time)
 
-            # Update history frequency for smart completion
-            if self._smart_completion_history and query.successful:
-                try:
-                    if hasattr(self.completer, 'update_history_from_query'):
-                        self.completer.update_history_from_query(text)
-                    elif hasattr(self.completer, '_history_tracker') and self.completer._history_tracker:
-                        # Fallback: record main keywords from the query
-                        import re
-                        import sqlparse
-                        try:
-                            parsed = sqlparse.parse(text)
-                            for statement in parsed:
-                                first_token = statement.token_first()
-                                if first_token:
-                                    keyword = str(first_token).upper().strip()
-                                    if keyword and not keyword.startswith('\\'):
-                                        self.completer._history_tracker.record_usage(keyword)
-                        except Exception:
-                            pass
-                except Exception as e:
-                    logger.debug("Error updating history frequency: %s", e)
-
             # Check if we need to update completions, in order of most
             # to least drastic changes
             if query.db_changed: