plugin_discover

class DiscoverResult:
    """Result of Plug-ins discovery of a single superclass type.

    Stores discovered, duplicated, ignored and abstract plugins and file paths
    which crashed on execution of file.
    """

    def __init__(self, superclass):
        self.superclass = superclass
        self.plugins = []
        self.crashed_file_paths = {}
        self.duplicated_plugins = []
        self.abstract_plugins = []
        self.ignored_plugins = set()
        # Store loaded modules to keep them in memory
        self._modules = set()

    def __iter__(self):
        for plugin in self.plugins:
            yield plugin

    def __getitem__(self, item):
        return self.plugins[item]

    def __setitem__(self, item, value):
        self.plugins[item] = value

    def add_module(self, module: ModuleType) -> None:
        """Add dynamically loaded python module to keep it in memory."""
        self._modules.add(module)

    def ignore_plugins(self, plugins: list[type]) -> None:
        """Add plugins to be ignored from the result."""
        for plugin in plugins:
            if plugin in self.ignored_plugins:
                continue
            if plugin in self.plugins:
                self.plugins.remove(plugin)
            self.ignored_plugins.add(plugin)

    def remove_duplicates(self) -> None:
        """Remove duplicated plugins from the result."""
        plugin_names: set[str] = set()
        for plugin in tuple(self.plugins):
            class_name = plugin.__name__
            if class_name in plugin_names:
                self.plugins.remove(plugin)
                self.duplicated_plugins.append(plugin)
                continue
            plugin_names.add(class_name)

    def get_report(
        self,
        only_errors: bool = True,
        exc_info: bool = True,
        full_report: bool = False,
    ) -> str:
        lines = []
        if not only_errors:
            # Successfully discovered plugins
            if self.plugins or full_report:
                lines.append(f"*** Discovered {len(self.plugins)} plugins")
                for cls in self.plugins:
                    lines.append(f"- {cls.__name__}")

            # Plugin that were defined to be ignored
            if self.ignored_plugins or full_report:
                lines.append(
                    f"*** Ignored plugins {len(self.ignored_plugins)}"
                )
                for cls in self.ignored_plugins:
                    lines.append(f"- {cls.__name__}")

        # Abstract classes
        if self.abstract_plugins or full_report:
            lines.append(
                f"*** Discovered {len(self.abstract_plugins)}"
                " abstract plugins"
            )
            for cls in self.abstract_plugins:
                lines.append(f"- {cls.__name__}")

        # Abstract classes
        if self.duplicated_plugins or full_report:
            lines.append(
                f"*** There were {len(self.duplicated_plugins)}"
                " duplicated plugins"
            )
            for cls in self.duplicated_plugins:
                lines.append(f"- {cls.__name__}")

        if self.crashed_file_paths or full_report:
            lines.append(
                f"*** Failed to load {len(self.crashed_file_paths)} files"
            )
            for path, exc_info_args in self.crashed_file_paths.items():
                lines.append(f"- {path}")
                if exc_info:
                    lines.append(10 * "*")
                    lines.extend(traceback.format_exception(*exc_info_args))
                    lines.append(10 * "*")

        return "\n".join(lines)

    def log_report(
        self, only_errors: bool = True, exc_info: bool = True,
    ) -> None:
        report = self.get_report(only_errors, exc_info)
        if report:
            log.info(report)

@dataclass
class DiscoveryDefinition:
    """Superclass discovery definition.

    Attributes:
        superclass (type): Superclass to discover subclasses in paths.
        paths (list[str]): List of paths to look for plug-ins that do inherit
            from the superclass.
        classes (list[type]): List of classes that are not discovered from
            paths but are explicitly defined. Should inherit from
                the superclass.

    """
    superclass: type
    paths: list[str] = field(default_factory=list)
    classes: list[type] = field(default_factory=list)

class PluginDiscoverContext:
    """Store and discover registered types nad registered paths to types.

    Keeps in memory all registered types and their paths. Paths are dynamically
    loaded on discover so different discover calls won't return the same
    class objects even if were loaded from same file.
    """

    def __init__(self):
        self._registered_plugins = {}
        self._registered_plugin_paths = {}
        self._last_discovered_plugins = {}
        # Store the last result to memory
        self._last_discovered_results = {}

    def get_last_discovered_plugins(self, superclass):
        """Access last discovered plugin by a subperclass.

        Returns:
            None: When superclass was not discovered yet.
            list: Lastly discovered plugins of the superclass.
        """

        return self._last_discovered_plugins.get(superclass)

    def discover(
        self,
        superclass,
        allow_duplicates=True,
        ignore_classes=None,
        return_report=False
    ):
        """Find and return subclasses of `superclass`

        Args:
            superclass (type): Class which determines discovered subclasses.
            allow_duplicates (bool): Validate class name duplications.
            ignore_classes (list): List of classes that will be ignored
                and not added to result.
            return_report (bool): Output will be full report if set to 'True'.

        Returns:
            Union[DiscoverResult, list[Any]]: Object holding successfully
                discovered plugins, ignored plugins, plugins with missing
                abstract implementation and duplicated plugin.

        """
        registered_classes = self._registered_plugins.get(superclass) or []
        registered_paths = self._registered_plugin_paths.get(superclass) or []
        result = discover_plugins(
            superclass,
            paths=registered_paths,
            classes=registered_classes,
            ignored_classes=ignore_classes,
            allow_duplicates=allow_duplicates,
        )

        # Store in memory last result to keep in memory loaded modules
        self._last_discovered_results[superclass] = result
        self._last_discovered_plugins[superclass] = list(
            result.plugins
        )
        result.log_report()
        if return_report:
            return result
        return result.plugins

    def discover_with_report(
        self, superclasses: Iterable[type]
    ) -> dict[type, DiscoverResult]:
        """Find and return subclasses of superclasses.

        Args:
            superclasses (Iterable[type]): Classes which determines discovered
                subclasses.

        Returns:
            dict[type, DiscoverResult]: Object holding successfully
                discovered plugins, ignored plugins, plugins with missing
                abstract implementation and duplicated plugin.

        """
        defs = [
            DiscoveryDefinition(
                superclass=superclass,
                paths=self._registered_plugin_paths.get(superclass) or [],
                classes=self._registered_plugins.get(superclass) or [],
            )
            for superclass in superclasses
        ]
        result = discover_plugins_with_defs(defs)

        # Store in memory last result to keep in memory loaded modules
        for superclass, sc_result in result.items():
            self._last_discovered_results[superclass] = sc_result
            self._last_discovered_plugins[superclass] = list(
                sc_result.plugins
            )
            sc_result.log_report()
        return result

    def register_plugin(self, superclass, cls):
        """Register a directory containing plug-ins of type `superclass`

        Arguments:
            superclass (type): Superclass of plug-in
            cls (object): Subclass of `superclass`

        """
        if superclass not in self._registered_plugins:
            self._registered_plugins[superclass] = list()

        if cls not in self._registered_plugins[superclass]:
            self._registered_plugins[superclass].append(cls)

    def register_plugin_path(self, superclass, path):
        """Register a directory of one or more plug-ins

        Arguments:
            superclass (type): Superclass of plug-ins to look for during
                discovery
            path (str): Absolute path to directory in which to discover
                plug-ins

        """
        if superclass not in self._registered_plugin_paths:
            self._registered_plugin_paths[superclass] = list()

        path = os.path.normpath(path)
        if path not in self._registered_plugin_paths[superclass]:
            self._registered_plugin_paths[superclass].append(path)

    def registered_plugin_paths(self):
        """Return all currently registered plug-in paths"""
        # Return shallow copy so we the original data can't be changed
        return {
            superclass: paths[:]
            for superclass, paths in self._registered_plugin_paths.items()
        }

    def deregister_plugin(self, superclass, plugin):
        """Opposite of `register_plugin()`"""
        if superclass in self._registered_plugins:
            self._registered_plugins[superclass].remove(plugin)

    def deregister_plugin_path(self, superclass, path):
        """Opposite of `register_plugin_path()`"""
        self._registered_plugin_paths[superclass].remove(path)