Class LazyInitEmitter

awscli/lazy_emitter.py:34–160

HierarchicalEmitter that lazily initializes plugins from a registry. The registry maps event patterns to lists of (module, fn_name) tuples. Before emitting any event, this emitter checks whether there are uninitialised plugins whose event patterns match the event being emitted. If s

class LazyInitEmitter(HierarchicalEmitter):
    """HierarchicalEmitter that lazily initializes plugins from a registry.

    The registry maps event patterns to lists of (module, fn_name) tuples.
    Before emitting any event, this emitter checks whether there are
    uninitialised plugins whose event patterns match the event being emitted.
    If so, it imports and calls them, then proceeds with normal dispatch.
    """

    def __init__(self, plugin_registry=None, main_command_table_ops=None):
        super().__init__()
        self._init_trie = PrefixTrie()
        # set of (module, fn_name)
        self._initialized = set()
        # number of entries not yet initialized
        self._pending_count = 0
        # event_name -> list of entries from init trie
        self._init_cache: dict[str, list] = {}
        self._registry = plugin_registry or {}
        self._main_ops = main_command_table_ops
        self._main_ops_applied = False
        if self._registry:
            self.load_registry(self._registry)

    def load_registry(self, registry):
        self._registry = registry
        unique = set()
        for event_pattern, entries in registry.items():
            for entry in entries:
                self._init_trie.append_item(event_pattern, entry)
                if entry not in unique:
                    unique.add(entry)
                    self._pending_count += 1
        self._init_cache = {}

    @property
    def initialized_count(self):
        return len(self._initialized)

    def _apply_main_command_table_ops(self, kwargs):
        """Apply pre-computed renames and LazyCommand additions.

        This replaces the normal lazy-init path for
        building-command-table.main entries, avoiding the import of
        heavy plugin modules until the command is actually invoked.
        """
        command_table = kwargs.get('command_table')
        session = kwargs.get('session')
        if command_table is None or session is None:
            return

        super()._emit('before-load-plugins.building-command-table.main', {})
        for op in self._main_ops:
            if op[0] == CommandTableOp.RENAME:
                _, old_name, new_name = op
                if old_name in command_table:
                    current = command_table[old_name]
                    command_table[new_name] = current