def get_plugin_paths()

in src/buildstream/_pluginfactory/pluginoriginpip.py [0:0]

• 58 lines of code
• 11 McCabe index (conditional complexity)

    def get_plugin_paths(self, kind, plugin_type):

        from packaging.requirements import Requirement, InvalidRequirement

        if sys.version_info >= (3, 10):
            from importlib.metadata import distribution, PackageNotFoundError
        else:
            from importlib_metadata import distribution, PackageNotFoundError

        # Sources and elements are looked up in separate
        # entrypoint groups from the same package.
        #
        if plugin_type == PluginType.SOURCE:
            entrypoint_group = "buildstream.plugins.sources"
        elif plugin_type == PluginType.ELEMENT:
            entrypoint_group = "buildstream.plugins.elements"
        elif plugin_type == PluginType.SOURCE_MIRROR:
            entrypoint_group = "buildstream.plugins.sourcemirrors"
        else:
            assert False, "unreachable"

        try:
            package = Requirement(self._package_name)
        except InvalidRequirement as e:
            raise PluginError(
                "{}: Malformed package-name '{}' encountered: {}".format(
                    self.provenance_node.get_provenance(), self._package_name, e
                ),
                reason="package-malformed-requirement",
            ) from e

        try:
            dist = distribution(package.name)
        except PackageNotFoundError as e:
            raise PluginError(
                "{}: Failed to load {} plugin '{}': {}".format(
                    self.provenance_node.get_provenance(), plugin_type, kind, e
                ),
                reason="package-not-found",
            ) from e

        if dist.version not in package.specifier:
            raise PluginError(
                "{}: Version conflict encountered while loading {} plugin '{}'".format(
                    self.provenance_node.get_provenance(), plugin_type, kind
                ),
                detail="{} {} is installed but {} is required".format(dist.name, dist.version, package),
                reason="package-version-conflict",
            )

        try:
            entrypoint = dist.entry_points.select(group=entrypoint_group)[kind]
        except KeyError as e:
            raise PluginError(
                "{}: Pip package {} does not contain a {} plugin named '{}'".format(
                    self.provenance_node.get_provenance(), self._package_name, plugin_type, kind
                ),
                reason="plugin-not-found",
            )

        location = dist.locate_file(entrypoint.module.replace(".", os.sep) + ".py")
        defaults = dist.locate_file(entrypoint.module.replace(".", os.sep) + ".yaml")

        if not defaults.exists():
            # The plugin didn't have an accompanying YAML file
            defaults = None

        return (
            os.path.dirname(location),
            str(defaults),
            "python package '{}' at: {}".format(dist, dist.locate_file("")),
        )