{Core} Use commandIndex caching to improve performance of top-level and module-level help commands

src/azure-cli-core/azure/cli/core/__init__.py

@@ -429,6 +429,28 @@ def _get_extension_suppressions(mod_loaders):
         command_index = None
         # Set fallback=False to turn off command index in case of regression
         use_command_index = self.cli_ctx.config.getboolean('core', 'use_command_index', fallback=True)
+
+        # Fast path for help requests - check if we can use cached help index to skip module loading
+        if use_command_index and args and ('--help' in args or '-h' in args or args[-1] == 'help'):
+            # Parse the command path from args (e.g., ['vm', '--help'] -> 'vm')
+            command_path_parts = []
+            for arg in args:
+                if arg in ('--help', '-h', 'help'):
+                    break
+                if not arg.startswith('-'):
+                    command_path_parts.append(arg)
+
+            command_path = ' '.join(command_path_parts) if command_path_parts else 'root'
+
+            command_index = CommandIndex(self.cli_ctx)
+            help_index = command_index.get_help_index()
+            if help_index and command_path in help_index:
+                logger.debug("Using cached help index for '%s', skipping module loading", command_path)
+                # Display help directly from cached data without loading modules
+                self._display_cached_help(help_index[command_path], command_path)
+                # Raise SystemExit to stop execution (similar to how --help normally works)
+                sys.exit(0)
+        # Fast path for top-level with no args (az with no arguments)
         if use_command_index:
             command_index = CommandIndex(self.cli_ctx)
             index_result = command_index.get(args)
@@ -496,8 +518,231 @@ def _get_extension_suppressions(mod_loaders):
 
         if use_command_index:
             command_index.update(self.command_table)
+
+            # Also cache help data for fast az --help in future
+            # This is done after loading all modules when help data is available
+            self._cache_help_index(command_index)
 
         return self.command_table
+
+    def _display_cached_help(self, help_data, command_path='root'):
+        """Display help from cached help index without loading modules."""
+        from azure.cli.core._help import WELCOME_MESSAGE, PRIVACY_STATEMENT
+        import re
+
+        def _strip_ansi(text):
+            """Remove ANSI color codes from text for length calculation."""
+            ansi_escape = re.compile(r'\x1b\[[0-9;]*m')
+            return ansi_escape.sub('', text)
+
+        # Show privacy statement if first run
+        ran_before = self.cli_ctx.config.getboolean('core', 'first_run', fallback=False)
+        if not ran_before:
+            print(PRIVACY_STATEMENT)
+            self.cli_ctx.config.set_value('core', 'first_run', 'yes')
+
+        # Show welcome message
+        print(WELCOME_MESSAGE)
+
+        # Display the group breadcrumb
+        if command_path == 'root':
+            print("\nGroup")
+            print("    az")
+        else:
+            print("\nGroup")
+            print(f"    az {command_path}")
+
+        # Import knack's formatting functions
+        from knack.help import _print_indent, FIRST_LINE_PREFIX, _get_hanging_indent
+
+        # Separate groups and commands
+        groups_data = help_data.get('groups', {})
+        commands_data = help_data.get('commands', {})
+
+        # Helper function matching knack's _get_line_len
+        def _get_line_len(name, tags):
+            tags_len = len(_strip_ansi(tags))
+            return len(name) + tags_len + (2 if tags_len else 1)
+
+        # Helper function matching knack's _get_padding_len
+        def _get_padding_len(max_len, name, tags):
+            line_len = _get_line_len(name, tags)
+            if tags:
+                pad_len = max_len - line_len + 1
+            else:
+                pad_len = max_len - line_len
+            return pad_len
+
+        # Build items lists and calculate max_line_len across ALL items (groups + commands)
+        # This ensures colons align across both sections
+        max_line_len = 0
+        groups_items = []
+        for name in sorted(groups_data.keys()):
+            item = groups_data[name]
+            tags = item.get('tags', '')
+            groups_items.append((name, tags, item.get('summary', '')))
+            max_line_len = max(max_line_len, _get_line_len(name, tags))
+
+        commands_items = []
+        for name in sorted(commands_data.keys()):
+            item = commands_data[name]
+            tags = item.get('tags', '')
+            commands_items.append((name, tags, item.get('summary', '')))
+            max_line_len = max(max_line_len, _get_line_len(name, tags))
+
+        # Display groups
+        if groups_items:
+            print("\nSubgroups:")
+            indent = 1
+            LINE_FORMAT = '{name}{padding}{tags}{separator}{summary}'
+            for name, tags, summary in groups_items:
+                padding = ' ' * _get_padding_len(max_line_len, name, tags)
+                line = LINE_FORMAT.format(
+                    name=name,
+                    padding=padding,
+                    tags=tags,
+                    separator=FIRST_LINE_PREFIX if summary else '',
+                    summary=summary
+                )
+                _print_indent(line, indent, _get_hanging_indent(max_line_len, indent))
+
+        # Display commands
+        if commands_items:
+            print("\nCommands:")
+            indent = 1
+            LINE_FORMAT = '{name}{padding}{tags}{separator}{summary}'
+            for name, tags, summary in commands_items:
+                padding = ' ' * _get_padding_len(max_line_len, name, tags)
+                line = LINE_FORMAT.format(
+                    name=name,
+                    padding=padding,
+                    tags=tags,
+                    separator=FIRST_LINE_PREFIX if summary else '',
+                    summary=summary
+                )
+                _print_indent(line, indent, _get_hanging_indent(max_line_len, indent))
+
+        print("\nTo search AI knowledge base for examples, use: az find \"az \"")
+
+        # Show update notification
+        from azure.cli.core.util import show_updates_available
+        show_updates_available(new_line_after=True)
+
+    def _cache_help_index(self, command_index):
+        """Cache help summaries for all commands/groups recursively using parallel processing."""
+        try:
+            import concurrent.futures
+            from concurrent.futures import ThreadPoolExecutor
+            from azure.cli.core.parser import AzCliCommandParser
+            from azure.cli.core._help import CliGroupHelpFile
+
+            # Create a temporary parser to extract help information
+            parser = AzCliCommandParser(self.cli_ctx)
+            parser.load_command_table(self)
+
+            # Helper to build tag string for an item
+            def _get_tags(item):
+                tags = []
+                if hasattr(item, 'deprecate_info') and item.deprecate_info:
+                    tags.append(str(item.deprecate_info.tag))
+                if hasattr(item, 'preview_info') and item.preview_info:
+                    tags.append(str(item.preview_info.tag))
+                if hasattr(item, 'experimental_info') and item.experimental_info:
+                    tags.append(str(item.experimental_info.tag))
+                return ' '.join(tags)
+
+            # Function to cache help for a single group
+            def _cache_single_group_help(group_path, subparser):
+                """Cache help for a single group and return its data along with subgroups to process."""
+                try:
+                    help_file = CliGroupHelpFile(self.cli_ctx.invocation.help, group_path, subparser)
+                    help_file.load(subparser)
+
+                    groups = {}
+                    commands = {}
+                    subgroup_names = []
+
+                    for child in help_file.children:
+                        if hasattr(child, 'name') and hasattr(child, 'short_summary'):
+                            # Extract just the last part of the name (after the group path)
+                            child_name = child.name
+                            if group_path and child_name.startswith(group_path + ' '):
+                                child_name = child_name[len(group_path) + 1:]
+                            elif not group_path and ' ' in child_name:
+                                # Skip nested items at root level
+                                continue
+
+                            tags = _get_tags(child)
+                            item_data = {
+                                'summary': child.short_summary,
+                                'tags': tags
+                            }
+
+                            if child.type == 'group':
+                                groups[child_name] = item_data
+                                # Build full path for recursion
+                                if group_path:
+                                    full_subgroup_name = f"{group_path} {child_name}"
+                                else:
+                                    full_subgroup_name = child_name
+                                subgroup_names.append(full_subgroup_name)
+                            else:
+                                commands[child_name] = item_data
+
+                    level_key = group_path if group_path else 'root'
+                    level_data = None
+                    if groups or commands:
+                        level_data = {'groups': groups, 'commands': commands}
+
+                    return level_key, level_data, subgroup_names
+
+                except Exception as ex:  # pylint: disable=broad-except
+                    logger.debug("Failed to cache help for '%s': %s", group_path, ex)
+                    return None, None, []
+
+            # Build help index using BFS with parallel processing at each level
+            help_index_data = {}
+            to_process = [('', parser.subparsers.get(tuple()))]  # (group_path, subparser) tuples
+
+            while to_process:
+                # Process current level in parallel
+                with ThreadPoolExecutor(max_workers=4) as executor:
+                    futures = {}
+                    for group_path, subparser in to_process:
+                        if subparser:
+                            future = executor.submit(_cache_single_group_help, group_path, subparser)
+                            futures[future] = group_path
+
+                    next_level = []
+                    for future in concurrent.futures.as_completed(futures):
+                        try:
+                            level_key, level_data, subgroup_names = future.result(timeout=10)
+                            if level_data:
+                                help_index_data[level_key] = level_data
+
+                            # Queue subgroups for next level
+                            for subgroup_name in subgroup_names:
+                                subgroup_tuple = tuple(subgroup_name.split())
+                                sub_subparser = parser.subparsers.get(subgroup_tuple)
+                                if sub_subparser:
+                                    next_level.append((subgroup_name, sub_subparser))
+
+                        except concurrent.futures.TimeoutError:
+                            group_path = futures[future]
+                            logger.debug("Help caching timeout for '%s'", group_path)
+                        except Exception as ex:  # pylint: disable=broad-except
+                            group_path = futures[future]
+                            logger.debug("Failed to cache help for '%s': %s", group_path, ex)
+
+                    to_process = next_level
+
+            # Store the complete help index in one operation
+            if help_index_data:
+                command_index.INDEX[command_index._HELP_INDEX] = help_index_data
+                logger.debug("Cached help for %d command levels", len(help_index_data))
+
+        except Exception as ex:  # pylint: disable=broad-except
+            logger.debug("Failed to cache help data: %s", ex)
 
     @staticmethod
     def _sort_command_loaders(command_loaders):
@@ -567,6 +812,7 @@ class CommandIndex:
     _COMMAND_INDEX = 'commandIndex'
     _COMMAND_INDEX_VERSION = 'version'
     _COMMAND_INDEX_CLOUD_PROFILE = 'cloudProfile'
+    _HELP_INDEX = 'helpIndex'
 
     def __init__(self, cli_ctx=None):
         """Class to manage command index.
@@ -635,6 +881,25 @@ def get(self, args):
 
         return None
 
+    def get_help_index(self):
+        """Get the help index for top-level help display.
+
+        :return: Dictionary mapping top-level commands to their short summaries, or None if not available
+        """
+        # Check if index is valid
+        index_version = self.INDEX[self._COMMAND_INDEX_VERSION]
+        cloud_profile = self.INDEX[self._COMMAND_INDEX_CLOUD_PROFILE]
+        if not (index_version and index_version == self.version and
+                cloud_profile and cloud_profile == self.cloud_profile):
+            return None
+
+        help_index = self.INDEX.get(self._HELP_INDEX, {})
+        if help_index:
+            logger.debug("Using cached help index with %d entries", len(help_index))
+            return help_index
+
+        return None
+
     def update(self, command_table):
         """Update the command index according to the given command table.
 
@@ -645,6 +910,7 @@ def update(self, command_table):
         self.INDEX[self._COMMAND_INDEX_CLOUD_PROFILE] = self.cloud_profile
         from collections import defaultdict
         index = defaultdict(list)
+        help_index = {}  # Maps top-level command to short summary
 
         # self.cli_ctx.invocation.commands_loader.command_table doesn't exist in DummyCli due to the lack of invocation
         for command_name, command in command_table.items():
@@ -654,8 +920,11 @@ def update(self, command_table):
             module_name = command.loader.__module__
             if module_name not in index[top_command]:
                 index[top_command].append(module_name)
+
         elapsed_time = timeit.default_timer() - start_time
         self.INDEX[self._COMMAND_INDEX] = index
+        # Note: helpIndex is populated separately when az --help is displayed
+        # We don't populate it here because the help data isn't available yet
         logger.debug("Updated command index in %.3f seconds.", elapsed_time)
 
     def invalidate(self):
@@ -672,6 +941,7 @@ def invalidate(self):
         self.INDEX[self._COMMAND_INDEX_VERSION] = ""
         self.INDEX[self._COMMAND_INDEX_CLOUD_PROFILE] = ""
         self.INDEX[self._COMMAND_INDEX] = {}
+        self.INDEX[self._HELP_INDEX] = {}
         logger.debug("Command index has been invalidated.")
 
 
@@ -967,3 +1237,4 @@ def get_default_cli():
                  logging_cls=AzCliLogging,
                  output_cls=AzOutputProducer,
                  help_cls=AzCliHelp)
+

src/azure-cli-core/azure/cli/core/commands/__init__.py

@@ -578,6 +578,58 @@ def execute(self, args):
             self.parser.enable_autocomplete()
             subparser = self.parser.subparsers[tuple()]
             self.help.show_welcome(subparser)
+
+            # After showing help, cache the help summaries for future fast access
+            # This allows subsequent `az --help` calls to skip module loading
+            use_command_index = self.cli_ctx.config.getboolean('core', 'use_command_index', fallback=True)
+            logger.debug("About to cache help data, use_command_index=%s", use_command_index)
+            if use_command_index:
+                try:
+                    from azure.cli.core import CommandIndex
+                    command_index = CommandIndex(self.cli_ctx)
+                    # Extract help data from the parser that was just used
+                    from azure.cli.core._help import CliGroupHelpFile
+                    help_file = CliGroupHelpFile(self.cli_ctx.help, '', subparser)
+                    help_file.load(subparser)
+
+                    # Helper to build tag string for an item
+                    def _get_tags(item):
+                        tags = []
+                        if hasattr(item, 'deprecate_info') and item.deprecate_info:
+                            tags.append(str(item.deprecate_info.tag))
+                        if hasattr(item, 'preview_info') and item.preview_info:
+                            tags.append(str(item.preview_info.tag))
+                        if hasattr(item, 'experimental_info') and item.experimental_info:
+                            tags.append(str(item.experimental_info.tag))
+                        return ' '.join(tags)
+
+                    # Separate groups and commands
+                    groups = {}
+                    commands = {}
+
+                    for child in help_file.children:
+                        if hasattr(child, 'name') and hasattr(child, 'short_summary'):
+                            if ' ' not in child.name:  # Only top-level items
+                                tags = _get_tags(child)
+                                item_data = {
+                                    'summary': child.short_summary,
+                                    'tags': tags
+                                }
+                                # Check if it's a group or command
+                                if child.type == 'group':
+                                    groups[child.name] = item_data
+                                else:
+                                    commands[child.name] = item_data
+
+                    # Store in the command index
+                    help_index_data = {'groups': groups, 'commands': commands}
+                    if groups or commands:
+                        from azure.cli.core._session import INDEX
+                        from azure.cli.core import __version__
+                        INDEX['helpIndex'] = help_index_data
+                        logger.debug("Cached %d groups and %d commands for fast access", len(groups), len(commands))
+                except Exception as ex:  # pylint: disable=broad-except
+                    logger.debug("Failed to cache help data: %s", ex)
 
             # TODO: No event in base with which to target
             telemetry.set_command_details('az', command_preserve_casing=command_preserve_casing)