5756139943493632

Committed 14 Mar 2025 03:41AM UTC coverage: 98.055% (+0.003%) from 98.052%

Build # 5756139943493632

Build Type

Pull #243

cirrus-ci

Committed by

web-flow

Commit Message

Remove expression difficult to understand

Pull Request Pull Request #243: Proposal: Alternative plugin sorting

Run Details

315 of 327 branches covered (96.33%)

Branch coverage included in aggregate %.

11 of 11 new or added lines in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

996 of 1010 relevant lines covered (98.61%)

6.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.51

/src/validate_pyproject/plugins/__init__.py

# The code in this module is mostly borrowed/adapted from PyScaffold and was originally
# published under the MIT license
# The original PyScaffold license can be found in 'NOTICE.txt'
"""
.. _entry point: https://setuptools.readthedocs.io/en/latest/userguide/entry_point.html
"""

import typing
from importlib.metadata import EntryPoint, entry_points
from itertools import chain
from string import Template
from textwrap import dedent
from typing import (
    Any,
    Callable,
    Generator,
    Iterable,
    List,
    NamedTuple,
    Optional,
    Protocol,
    Union,
)

from .. import __version__
from ..types import Plugin, Schema


class PluginProtocol(Protocol):
    @property
    def id(self) -> str: ...

    @property
    def tool(self) -> str: ...

    @property
    def schema(self) -> Schema: ...

    @property
    def help_text(self) -> str: ...

    @property
    def fragment(self) -> str: ...


class PluginWrapper:
    def __init__(self, tool: str, load_fn: Plugin):
        self._tool = tool
        self._load_fn = load_fn

    @property
    def id(self) -> str:
        return f"{self._load_fn.__module__}.{self._load_fn.__name__}"

    @property
    def tool(self) -> str:
        return self._tool

    @property
    def schema(self) -> Schema:
        return self._load_fn(self.tool)

    @property
    def fragment(self) -> str:
        return ""

    @property
    def help_text(self) -> str:
        tpl = self._load_fn.__doc__
        if not tpl:
            return ""
        return Template(tpl).safe_substitute(tool=self.tool, id=self.id)

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}({self.tool!r}, {self.id})"

    def __str__(self) -> str:
        return self.id


class StoredPlugin:
    def __init__(self, tool: str, schema: Schema, source: str):
        self._tool, _, self._fragment = tool.partition("#")
        self._schema = schema
        self._source = source

    @property
    def id(self) -> str:
        return self._schema["$id"]  # type: ignore[no-any-return]

    @property
    def tool(self) -> str:
        return self._tool

    @property
    def schema(self) -> Schema:
        return self._schema

    @property
    def fragment(self) -> str:
        return self._fragment

    @property
    def help_text(self) -> str:
        return self.schema.get("description", "")

    def __str__(self) -> str:
        return self._source

    def __repr__(self) -> str:
        args = [repr(self.tool), self.id]
        if self.fragment:
            args.append(f"fragment={self.fragment!r}")
        return f"{self.__class__.__name__}({', '.join(args)}, <schema: {self.id}>)"


if typing.TYPE_CHECKING:
    _: PluginProtocol = typing.cast(PluginWrapper, None)


def iterate_entry_points(group: str) -> Iterable[EntryPoint]:
    """Produces an iterable yielding an EntryPoint object for each plugin registered
    via ``setuptools`` `entry point`_ mechanism.

    This method can be used in conjunction with :obj:`load_from_entry_point` to filter
    the plugins before actually loading them. The entry points are not
    deduplicated.
    """
    entries = entry_points()
    if hasattr(entries, "select"):  # pragma: no cover
        # The select method was introduced in importlib_metadata 3.9 (and Python 3.10)
        # and the previous dict interface was declared deprecated
        select = typing.cast(
            Callable[..., Iterable[EntryPoint]],
            getattr(entries, "select"),  # noqa: B009
        )  # typecheck gymnastics
        return select(group=group)
    # pragma: no cover
    # TODO: Once Python 3.10 becomes the oldest version supported, this fallback and
    #       conditional statement can be removed.
    return (plugin for plugin in entries.get(group, []))


def load_from_entry_point(entry_point: EntryPoint) -> PluginWrapper:
    """Carefully load the plugin, raising a meaningful message in case of errors"""
    try:
        fn = entry_point.load()
        return PluginWrapper(entry_point.name, fn)
    except Exception as ex:
        raise ErrorLoadingPlugin(entry_point=entry_point) from ex


def load_from_multi_entry_point(
    entry_point: EntryPoint,
) -> Generator[StoredPlugin, None, None]:
    """Carefully load the plugin, raising a meaningful message in case of errors"""
    try:
        fn = entry_point.load()
        output = fn()
        id_ = f"{fn.__module__}.{fn.__name__}"
    except Exception as ex:
        raise ErrorLoadingPlugin(entry_point=entry_point) from ex

    for tool, schema in output["tools"].items():
        yield StoredPlugin(tool, schema, f"{id_}:{tool}")
    for i, schema in enumerate(output.get("schemas", [])):
        yield StoredPlugin("", schema, f"{id_}:{i}")


class _SortablePlugin(NamedTuple):
    priority: int
    name: str
    plugin: Union[PluginWrapper, StoredPlugin]

    def key(self) -> str:
        return self.plugin.tool or self.plugin.id

    def __lt__(self, other: Any) -> bool:
        return (self.priority, self.name, self.key()) < (
            other.priority,
            other.name,
            other.key(),
        )


def list_from_entry_points(
    filtering: Callable[[EntryPoint], bool] = lambda _: True,
) -> List[Union[PluginWrapper, StoredPlugin]]:
    """Produces a list of plugin objects for each plugin registered
    via ``setuptools`` `entry point`_ mechanism.

    Args:
        filtering: function returning a boolean deciding if the entry point should be
            loaded and included (or not) in the final list. A ``True`` return means the
            plugin should be included.
    """
    # **Major concern**:
    # Consistency and reproducibility on which entry-points have priority
    # for a given environment.
    # The plugin with higher priority overwrites the schema definition.
    # The exact order itself is not important for now.
    # **Implementation detail**:
    # Tool plugins are loaded first, so they are listed first than other schemas,
    # but multi plugins always have priority, overwriting the tool schemas.
    # The "higher alphabetically" an entry-point name, the more priority.
    tool_eps = (
        _SortablePlugin(0, e.name, load_from_entry_point(e))
        for e in iterate_entry_points("validate_pyproject.tool_schema")
        if filtering(e)
    )
    multi_eps = (
        _SortablePlugin(1, e.name, p)
        for e in iterate_entry_points("validate_pyproject.multi_schema")
        for p in load_from_multi_entry_point(e)
        if filtering(e)
    )
    eps = chain(tool_eps, multi_eps)
    dedup = {e.key(): e.plugin for e in sorted(eps)}
    return list(dedup.values())


class ErrorLoadingPlugin(RuntimeError):
    _DESC = """There was an error loading '{plugin}'.
    Please make sure you have installed a version of the plugin that is compatible
    with {package} {version}. You can also try uninstalling it.
    """
    __doc__ = _DESC

    def __init__(self, plugin: str = "", entry_point: Optional[EntryPoint] = None):
        if entry_point and not plugin:
            plugin = getattr(entry_point, "module", entry_point.name)

        sub = {"package": __package__, "version": __version__, "plugin": plugin}
        msg = dedent(self._DESC).format(**sub).splitlines()
        super().__init__(f"{msg[0]}\n{' '.join(msg[1:])}")

1	# The code in this module is mostly borrowed/adapted from PyScaffold and was originally
2	# published under the MIT license
3	# The original PyScaffold license can be found in 'NOTICE.txt'
4	"""	7✔
5	.. _entry point: https://setuptools.readthedocs.io/en/latest/userguide/entry_point.html
6	"""
7
8	import typing	7✔
9	from importlib.metadata import EntryPoint, entry_points	7✔
10	from itertools import chain	7✔
11	from string import Template	7✔
12	from textwrap import dedent	7✔
13	from typing import (	7✔
14	Any,
15	Callable,
16	Generator,
17	Iterable,
18	List,
19	NamedTuple,
20	Optional,
21	Protocol,
22	Union,
23	)
24
25	from .. import __version__	7✔
26	from ..types import Plugin, Schema	7✔
27
28
29	class PluginProtocol(Protocol):	7✔
30	@property	7✔
31	def id(self) -> str: ...	7!
32
33	@property	7✔
34	def tool(self) -> str: ...	7!
35
36	@property	7✔
37	def schema(self) -> Schema: ...	7!
38
39	@property	7✔
40	def help_text(self) -> str: ...	7!
41
42	@property	7✔
43	def fragment(self) -> str: ...	7!
44
45
46	class PluginWrapper:	7✔
47	def __init__(self, tool: str, load_fn: Plugin):	7✔
48	self._tool = tool	7✔
49	self._load_fn = load_fn	7✔
50
51	@property	7✔
52	def id(self) -> str:	7✔
53	return f"{self._load_fn.__module__}.{self._load_fn.__name__}"	7✔
54
55	@property	7✔
56	def tool(self) -> str:	7✔
57	return self._tool	7✔
58
59	@property	7✔
60	def schema(self) -> Schema:	7✔
61	return self._load_fn(self.tool)	7✔
62
63	@property	7✔
64	def fragment(self) -> str:	7✔
65	return ""	7✔
66
67	@property	7✔
68	def help_text(self) -> str:	7✔
69	tpl = self._load_fn.__doc__	7✔
70	if not tpl:	7✔
71	return ""	7✔
72	return Template(tpl).safe_substitute(tool=self.tool, id=self.id)	7✔
73
74	def __repr__(self) -> str:
75	return f"{self.__class__.__name__}({self.tool!r}, {self.id})"
76
77	def __str__(self) -> str:	7✔
78	return self.id	7✔
79
80
81	class StoredPlugin:	7✔
82	def __init__(self, tool: str, schema: Schema, source: str):	7✔
83	self._tool, _, self._fragment = tool.partition("#")	7✔
84	self._schema = schema	7✔
85	self._source = source	7✔
86
87	@property	7✔
88	def id(self) -> str:	7✔
89	return self._schema["$id"] # type: ignore[no-any-return]	7✔
90
91	@property	7✔
92	def tool(self) -> str:	7✔
93	return self._tool	7✔
94
95	@property	7✔
96	def schema(self) -> Schema:	7✔
97	return self._schema	7✔
98
99	@property	7✔
100	def fragment(self) -> str:	7✔
101	return self._fragment	7✔
102
103	@property	7✔
104	def help_text(self) -> str:	7✔
105	return self.schema.get("description", "")	7✔
106
107	def __str__(self) -> str:	7✔
UNCOV 108	return self._source	×
109
110	def __repr__(self) -> str:
111	args = [repr(self.tool), self.id]
112	if self.fragment:
113	args.append(f"fragment={self.fragment!r}")
114	return f"{self.__class__.__name__}({', '.join(args)}, <schema: {self.id}>)"
115
116
117	if typing.TYPE_CHECKING:
118	_: PluginProtocol = typing.cast(PluginWrapper, None)
119
120
121	def iterate_entry_points(group: str) -> Iterable[EntryPoint]:	7✔
122	"""Produces an iterable yielding an EntryPoint object for each plugin registered
123	via ``setuptools`` `entry point`_ mechanism.
124
125	This method can be used in conjunction with :obj:`load_from_entry_point` to filter
126	the plugins before actually loading them. The entry points are not
127	deduplicated.
128	"""
129	entries = entry_points()	7✔
130	if hasattr(entries, "select"): # pragma: no cover
131	# The select method was introduced in importlib_metadata 3.9 (and Python 3.10)
132	# and the previous dict interface was declared deprecated
133	select = typing.cast(
134	Callable[..., Iterable[EntryPoint]],
135	getattr(entries, "select"), # noqa: B009
136	) # typecheck gymnastics
137	return select(group=group)
138	# pragma: no cover
139	# TODO: Once Python 3.10 becomes the oldest version supported, this fallback and
140	# conditional statement can be removed.
141	return (plugin for plugin in entries.get(group, []))	1✔
142
143
144	def load_from_entry_point(entry_point: EntryPoint) -> PluginWrapper:	7✔
145	"""Carefully load the plugin, raising a meaningful message in case of errors"""
146	try:	7✔
147	fn = entry_point.load()	7✔
148	return PluginWrapper(entry_point.name, fn)	7✔
149	except Exception as ex:	7✔
150	raise ErrorLoadingPlugin(entry_point=entry_point) from ex	7✔
151
152
153	def load_from_multi_entry_point(	7✔
154	entry_point: EntryPoint,
155	) -> Generator[StoredPlugin, None, None]:
156	"""Carefully load the plugin, raising a meaningful message in case of errors"""
157	try:	7✔
158	fn = entry_point.load()	7✔
159	output = fn()	7✔
160	id_ = f"{fn.__module__}.{fn.__name__}"	7✔
161	except Exception as ex:	7✔
162	raise ErrorLoadingPlugin(entry_point=entry_point) from ex	7✔
163
164	for tool, schema in output["tools"].items():	7✔
165	yield StoredPlugin(tool, schema, f"{id_}:{tool}")	7✔
166	for i, schema in enumerate(output.get("schemas", [])):	7✔
167	yield StoredPlugin("", schema, f"{id_}:{i}")	7✔
168
169
170	class _SortablePlugin(NamedTuple):	7✔
171	priority: int	7✔
172	name: str	7✔
173	plugin: Union[PluginWrapper, StoredPlugin]	7✔
174
175	def key(self) -> str:	7✔
176	return self.plugin.tool or self.plugin.id	7✔
177
178	def __lt__(self, other: Any) -> bool:	7✔
179	return (self.priority, self.name, self.key()) < (	7✔
180	other.priority,
181	other.name,
182	other.key(),
183	)
184
185
186	def list_from_entry_points(	7✔
187	filtering: Callable[[EntryPoint], bool] = lambda _: True,
188	) -> List[Union[PluginWrapper, StoredPlugin]]:
189	"""Produces a list of plugin objects for each plugin registered
190	via ``setuptools`` `entry point`_ mechanism.
191
192	Args:
193	filtering: function returning a boolean deciding if the entry point should be
194	loaded and included (or not) in the final list. A ``True`` return means the
195	plugin should be included.
196	"""
197	# Major concern:
198	# Consistency and reproducibility on which entry-points have priority
199	# for a given environment.
200	# The plugin with higher priority overwrites the schema definition.
201	# The exact order itself is not important for now.
202	# Implementation detail:
203	# Tool plugins are loaded first, so they are listed first than other schemas,
204	# but multi plugins always have priority, overwriting the tool schemas.
205	# The "higher alphabetically" an entry-point name, the more priority.
206	tool_eps = (	7✔
207	_SortablePlugin(0, e.name, load_from_entry_point(e))
208	for e in iterate_entry_points("validate_pyproject.tool_schema")
209	if filtering(e)
210	)
211	multi_eps = (	7✔
212	_SortablePlugin(1, e.name, p)
213	for e in iterate_entry_points("validate_pyproject.multi_schema")
214	for p in load_from_multi_entry_point(e)
215	if filtering(e)
216	)
217	eps = chain(tool_eps, multi_eps)	7✔
218	dedup = {e.key(): e.plugin for e in sorted(eps)}	7✔
219	return list(dedup.values())	7✔
220
221
222	class ErrorLoadingPlugin(RuntimeError):	7✔
223	_DESC = """There was an error loading '{plugin}'.	7✔
224	Please make sure you have installed a version of the plugin that is compatible
225	with {package} {version}. You can also try uninstalling it.
226	"""
227	__doc__ = _DESC	7✔
228
229	def __init__(self, plugin: str = "", entry_point: Optional[EntryPoint] = None):	7✔
230	if entry_point and not plugin:	7!
231	plugin = getattr(entry_point, "module", entry_point.name)	7✔
232
233	sub = {"package": __package__, "version": __version__, "plugin": plugin}	7✔
234	msg = dedent(self._DESC).format(**sub).splitlines()	7✔
235	super().__init__(f"{msg[0]}\n{' '.join(msg[1:])}")	7✔

abravalheri / validate-pyproject / 5756139943493632

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous