Add alternative favorite query interface /favorite

changelog.md

@@ -1,6 +1,11 @@
 Upcoming (TBD)
 ==============
 
+Features
+---------
+* Add alternative interface `/favorite` for favorite queries.
+
+
 Bug Fixes
 ---------
 * Lower Boundary tunnel stabilization pause to 0.15 sec.

mycli/TIPS

@@ -64,6 +64,8 @@ edit a query in an external editor using /edit <filename>!
 
 Favorite queries support Jinja templates and named arguments!
 
+Manage favorite queries with /favorite!
+
 /l lists databases!
 
 /once <filename> appends the next result to <filename>!

mycli/clibuffer.py

@@ -29,7 +29,7 @@ def _multiline_exception(text: str) -> bool:
     # Multi-statement favorite query is a special case. Because there will
     # be a semicolon separating statements, we can't consider semicolon an
     # EOL. Let's consider an empty line an EOL instead.
-    if first_word.startswith(("\\fs", "/fs")):
+    if iocommands.is_favorite_save_command(text):
         return orig.endswith("\n")
 
     return (

mycli/client.py

@@ -122,6 +122,7 @@ def __init__(
             self.config,
             myclirc,
             c['main'].get('shared_favorites_file'),
+            system_config_files=self.system_config_files,
         )
         DsnAliases.instance = DsnAliases.from_config(
             self.config,

mycli/config.py

@@ -48,6 +48,7 @@ def read_config_file(
     f: str | IO[str],
     list_values: bool = True,
     preserve_quotes: bool = False,
+    raise_errors: bool = False,
 ) -> ConfigObj | LimiitedQuotePreservingConfigObj | None:
     """Read a config file.
 
@@ -57,6 +58,9 @@ def read_config_file(
     not unquoted. We are disabling list_values when reading MySQL config files
     so we can correctly interpret commas in passwords.
 
+    Set *raise_errors* to propagate parsing and I/O errors instead of logging
+    them and returning a partial config or ``None``.
+
     """
 
     if isinstance(f, str):
@@ -68,10 +72,14 @@ def read_config_file(
         else:
             config = ConfigObj(f, interpolation=False, encoding="utf8", list_values=list_values)
     except ConfigObjError as e:
+        if raise_errors:
+            raise
         log(logger, logging.WARNING, "Unable to parse line {0} of config file '{1}'.".format(e.line_number, f))
         log(logger, logging.WARNING, "Using successfully parsed config values.")
         return e.config
     except (IOError, OSError) as e:
+        if raise_errors:
+            raise
         log(logger, logging.WARNING, "You don't have permission to read config file '{0}'.".format(e.filename))
         return None
 

mycli/main_modes/repl.py

@@ -105,6 +105,7 @@
 class ReplState:
     iterations: int = 0
     mutating: bool = False
+    buffer_text: str | None = None
 
 
 @Condition
@@ -449,6 +450,9 @@ def _output_results(
         mycli.logger.debug('status: %r', result.status)
         mycli.logger.debug('command: %r', result.command)
         threshold = 1000
+        if result.command is not None and result.command['name'] == 'set_buffer':
+            state.buffer_text = str(result.command['text'])
+            continue
         if result.command is not None and result.command['name'] == 'watch':
             if watch_count > 0:
                 try:
@@ -630,10 +634,19 @@ def _one_iteration(
         try:
             assert mycli.prompt_session is not None
             loaded_message_fn = partial(_get_prompt_message, mycli, mycli.prompt_session.app)
-            text = mycli.prompt_session.prompt(
-                inputhook=inputhook,
-                message=loaded_message_fn,
-            )
+            if state.buffer_text is None:
+                text = mycli.prompt_session.prompt(
+                    inputhook=inputhook,
+                    message=loaded_message_fn,
+                )
+            else:
+                default = state.buffer_text
+                state.buffer_text = None
+                text = mycli.prompt_session.prompt(
+                    default=default,
+                    inputhook=inputhook,
+                    message=loaded_message_fn,
+                )
         except KeyboardInterrupt:
             return
 

mycli/packages/completion_engine.py

@@ -8,6 +8,7 @@
 from sqlparse.sql import Comparison, Identifier, Token, Where
 
 from mycli.packages.special.dsn_aliases import DSN_SUBCOMMANDS
+from mycli.packages.special.favoritequeries import FAVORITE_SUBCOMMANDS
 from mycli.packages.special.main import COMMANDS as SPECIAL_COMMANDS
 from mycli.packages.special.main import parse_special_command
 from mycli.packages.sql_utils import extract_tables, find_prev_keyword, last_word
@@ -831,6 +832,36 @@ def suggest_special(text: str) -> list[dict[str, Any]]:
     if cmd in ["\\llm", "/llm", "\\ai", "/ai"]:
         return [{"type": "llm"}]
 
+    if cmd.lower() in (r'\favorite', '/favorite'):
+        favorite_arguments = _arg.split(maxsplit=1)
+        if favorite_arguments and favorite_arguments[0].lower() in ('run', 'eval'):
+            if len(favorite_arguments) == 1 and not text[-1].isspace():
+                return []
+            expansion_arg = favorite_arguments[1] if len(favorite_arguments) == 2 else ''
+            return suggest_favorite_query_with_template(text, expansion_arg)
+        if favorite_arguments and favorite_arguments[0].lower() == 'save':
+            if len(favorite_arguments) == 1 and not text[-1].isspace():
+                return []
+            if len(favorite_arguments) == 1:
+                return [{'type': 'favoritequery'}]
+            save_arguments = favorite_arguments[1].split(maxsplit=1)
+            if len(save_arguments) == 1 and not text[-1].isspace():
+                return [{'type': 'favoritequery'}]
+            query = save_arguments[1] if len(save_arguments) == 2 else ''
+            if query and text[-1].isspace():
+                query += ' '
+            return suggest_type(query, query)
+        if favorite_arguments and favorite_arguments[0].lower() in ('edit', 'delete'):
+            if len(favorite_arguments) == 1:
+                return [] if not text[-1].isspace() else [{'type': 'favoritequery'}]
+            target_arguments = favorite_arguments[1].split()
+            if len(target_arguments) == 1 and not text[-1].isspace():
+                return [{'type': 'favoritequery'}]
+            return []
+        if favorite_arguments and favorite_arguments[0].lower() in FAVORITE_SUBCOMMANDS:
+            return []
+        return [{'type': 'special_subcommand', 'subcommands': list(FAVORITE_SUBCOMMANDS)}]
+
     if cmd.lower() in (r'\config', '/config'):
         config_arguments = _arg.split(maxsplit=1)
         config_subcommands = ['help', 'get', 'search', 'edit']

mycli/packages/special/favoritequeries.py

@@ -1,23 +1,85 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 import logging
 import os
 import re
-from typing import Any
+from typing import IO, Any
 
+from configobj import ConfigObjError
 from jinja2 import meta, nodes
 from jinja2.sandbox import SandboxedEnvironment
 
-from mycli.config import log, read_config_file
+from mycli.config import log, read_config_file, read_config_files
 
 logger = logging.getLogger(__name__)
 
 MISSING = object()
+FAVORITE_SUBCOMMANDS = ('help', 'list', 'reload', 'run', 'eval', 'save', 'edit', 'delete')
+FAVORITE_COMMAND_HELP = '''
+Favorite Queries are a way to save frequently used queries
+with a short name.
+Examples:
+
+    # Save a new favorite query.
+    > /favorite save simple SELECT * FROM abc WHERE a IS NOT NULL;
+
+    # When multi-line mode is on, pressing Return twice is needed to save.
+    # This supports multi-statement favorites.
+
+    # List all favorite queries.
+    > /favorite list
+    ╒═══════════╤══════════════════════════════════════════════════╕
+    │ Name      │ Query                                            │
+    ╞═══════════╪══════════════════════════════════════════════════╡
+    │ simple    │ SELECT * FROM abc WHERE a IS NOT NULL            │
+    │ find_user │ SELECT * FROM users WHERE name = '{{ kv.name }}' │
+    ╘═══════════╧══════════════════════════════════════════════════╛
+
+    # Run a favorite query.
+    > /favorite run simple
+    ╒════════╤════════╕
+    │ a      │ b      │
+    ╞════════╪════════╡
+    │ 日本語 │ 日本語 │
+    ╘════════╧════════╛
+
+    # Run a favorite query containing {{ kv.name }} in the template.
+    > /favorite run find_user --name=henry
+    > /favorite run find_user --name henry
+
+    # Run a favorite query containing positional parameter $1 in the
+    # template.
+    > /favorite run find_user henry
+
+    # Use -- to disambiguate positional parameters such as $1, especially
+    # if the positional value starts with a dash.
+    > /favorite run query --key=value -- positional-value
+    > /favorite run query -- --positional-value-which-looks-like-a-flag--
+
+    # Expand a favorite query into the command-line buffer without running it.
+    > /favorite eval find_user --name=henry
+
+    # Edit a favorite query in an external editor.
+    > /favorite edit simple
+
+    # Delete a favorite query.
+    > /favorite delete simple
+    simple: Deleted.
+
+    # Reload favorite queries from the configuration files.
+    > /favorite reload
+
+    See also the alternative interface /f, /fs, /fd.'''
 
 favorite_query_template_environment = SandboxedEnvironment(autoescape=False)
 favorite_query_variable_pattern = re.compile(r'^[A-Za-z_][A-Za-z0-9_-]*$')
 
 
+class FavoriteQueryReloadError(Exception):
+    pass
+
+
 def analyze_favorite_query_template(query: str) -> tuple[set[str], bool]:
     """Return statically referenced keys and whether ``kv`` is used dynamically."""
     parsed_template = favorite_query_template_environment.parse(query)
@@ -116,25 +178,35 @@ class FavoriteQueries:
     # Delete a favorite query.
     > /fd simple
     simple: Deleted.
+
+    See also the alternative interface /favorite.
 """
 
     # Class-level variable, for convenience to use as a singleton.
     instance: FavoriteQueries
 
-    def __init__(self, config: Any, config_file: str | None = None) -> None:
+    def __init__(
+        self,
+        config: Any,
+        config_file: str | None = None,
+        shared_favorites_file: str | None = None,
+        system_config_files: list[str | IO[str]] | None = None,
+    ) -> None:
         self.config = config
         self.config_file = config_file
+        self.shared_favorites_file = shared_favorites_file
+        self.system_config_files = list(system_config_files or [])
 
     @classmethod
     def from_config(
         cls,
         config: Any,
         config_file: str | None = None,
         shared_favorites_file: str | None = None,
+        system_config_files: list[str | IO[str]] | None = None,
     ) -> FavoriteQueries:
-        favorites = cls(config, config_file)
         if not shared_favorites_file:
-            return favorites
+            return cls(config, config_file, system_config_files=system_config_files)
 
         shared_favorites_file = os.path.expanduser(shared_favorites_file)
         if not os.path.isabs(shared_favorites_file):
@@ -143,7 +215,9 @@ def from_config(
                 logging.WARNING,
                 f"Shared favorites file path must be absolute: '{shared_favorites_file}'.",
             )
-            return favorites
+            return cls(config, config_file, system_config_files=system_config_files)
+
+        favorites = cls(config, config_file, shared_favorites_file, system_config_files)
 
         if not os.path.isfile(shared_favorites_file):
             log(
@@ -164,6 +238,45 @@ def from_config(
         config[cls.section_name].update(configured_queries)
         return favorites
 
+    def _reload_queries(self, path: str, description: str) -> dict[str, str]:
+        expanded_path = os.path.expanduser(path)
+        if not os.path.isfile(expanded_path):
+            raise FavoriteQueryReloadError(f"unable to read {description} file '{expanded_path}'")
+        try:
+            config = read_config_file(expanded_path, raise_errors=True)
+        except (ConfigObjError, OSError, UnicodeError) as exc:
+            raise FavoriteQueryReloadError(
+                f"unable to read {description} file '{expanded_path}': {exc}",
+            ) from exc
+
+        assert config is not None
+        configured_queries = config.get(self.section_name, {})
+        if not isinstance(configured_queries, Mapping) or any(
+            not isinstance(name, str) or not isinstance(query, str) for name, query in configured_queries.items()
+        ):
+            raise FavoriteQueryReloadError(
+                f"invalid [{self.section_name}] section in {description} file '{expanded_path}'",
+            )
+        return dict(configured_queries)
+
+    def reload(self) -> None:
+        if self.config_file is None:
+            raise FavoriteQueryReloadError('no user configuration file is configured')
+
+        user_queries = self._reload_queries(self.config_file, 'user configuration')
+        system_config = read_config_files(self.system_config_files, ignore_package_defaults=True)
+        system_queries = system_config.get(self.section_name, {})
+        if not isinstance(system_queries, Mapping) or any(
+            not isinstance(name, str) or not isinstance(query, str) for name, query in system_queries.items()
+        ):
+            raise FavoriteQueryReloadError(f'invalid [{self.section_name}] section in system configuration files')
+        queries: dict[str, str] = {}
+        if self.shared_favorites_file is not None:
+            queries.update(self._reload_queries(self.shared_favorites_file, 'shared favorites'))
+        queries.update(system_queries)
+        queries.update(user_queries)
+        self.config[self.section_name] = queries
+
     def _clean_query(self, query: str | None) -> str | None:
         if not query:
             return query