6380044383158272

Committed 26 Feb 2025 08:12PM UTC coverage: 97.876% (-0.2%) from 98.033%

Build # 6380044383158272

Build Type

Pull #231

cirrus-ci

Committed by

henryiii

Commit Message

feat: support fragments

Signed-off-by: Henry Schreiner <henryschreineriii@gmail.com>

Pull Request Pull Request #231: feat: multi-plugins with extra schemas

Run Details

312 of 324 branches covered (96.3%)

Branch coverage included in aggregate %.

45 of 48 new or added lines in 4 files covered. (93.75%)

2 existing lines in 1 file now uncovered.

978 of 994 relevant lines covered (98.39%)

6.88 hits per line

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

93.48

/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 string import Template
from textwrap import dedent
from typing import Any, Callable, Generator, Iterable, List, 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})"


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

    @property
    def id(self) -> str:
        return self.schema.get("id", "MISSING ID")

    @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 __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 a generator 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, but they are sorted.
    """
    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(
            Any,
            getattr(entries, "select"),  # noqa: B009
        )  # typecheck gymnastics
        entries_: Iterable[EntryPoint] = select(group=group)
    else:  # pragma: no cover
        # TODO: Once Python 3.10 becomes the oldest version supported, this fallback and
        #       conditional statement can be removed.
        entries_ = (plugin for plugin in entries.get(group, []))
    return sorted(entries_, key=lambda e: e.name)


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()
    except Exception as ex:
        raise ErrorLoadingPlugin(entry_point=entry_point) from ex

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


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.
    """
    eps: List[Union[PluginWrapper, StoredPlugin]] = [
        load_from_entry_point(e)
        for e in iterate_entry_points("validate_pyproject.tool_schema")
        if filtering(e)
    ]
    for e in iterate_entry_points("validate_pyproject.multi_schema"):
        eps.extend(load_from_multi_entry_point(e))
    dedup = {(e.tool if e.tool else e.id): e for e in sorted(eps, key=lambda e: e.tool)}
    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 string import Template	7✔
11	from textwrap import dedent	7✔
12	from typing import Any, Callable, Generator, Iterable, List, Optional, Protocol, Union	7✔
13
14	from .. import __version__	7✔
15	from ..types import Plugin, Schema	7✔
16
17
18	class PluginProtocol(Protocol):	7✔
19	@property	7✔
20	def id(self) -> str: ...	7!
21
22	@property	7✔
23	def tool(self) -> str: ...	7!
24
25	@property	7✔
26	def schema(self) -> Schema: ...	7!
27
28	@property	7✔
29	def help_text(self) -> str: ...	7!
30
31	@property	7✔
32	def fragment(self) -> str: ...	7!
33
34
35	class PluginWrapper:	7✔
36	def __init__(self, tool: str, load_fn: Plugin):	7✔
37	self._tool = tool	7✔
38	self._load_fn = load_fn	7✔
39
40	@property	7✔
41	def id(self) -> str:	7✔
42	return f"{self._load_fn.__module__}.{self._load_fn.__name__}"	7✔
43
44	@property	7✔
45	def tool(self) -> str:	7✔
46	return self._tool	7✔
47
48	@property	7✔
49	def schema(self) -> Schema:	7✔
50	return self._load_fn(self.tool)	7✔
51
52	@property	7✔
53	def fragment(self) -> str:	7✔
54	return ""	7✔
55
56	@property	7✔
57	def help_text(self) -> str:	7✔
58	tpl = self._load_fn.__doc__	7✔
59	if not tpl:	7✔
60	return ""	7✔
61	return Template(tpl).safe_substitute(tool=self.tool, id=self.id)	7✔
62
63	def __repr__(self) -> str:
64	return f"{self.__class__.__name__}({self.tool!r}, {self.id})"
65
66
67	class StoredPlugin:	7✔
68	def __init__(self, tool: str, schema: Schema):	7✔
69	self._tool, _, self._fragment = tool.partition("#")	7✔
70	self._schema = schema	7✔
71
72	@property	7✔
73	def id(self) -> str:	7✔
74	return self.schema.get("id", "MISSING ID")	7✔
75
76	@property	7✔
77	def tool(self) -> str:	7✔
78	return self._tool	7✔
79
80	@property	7✔
81	def schema(self) -> Schema:	7✔
82	return self._schema	7✔
83
84	@property	7✔
85	def fragment(self) -> str:	7✔
86	return self._fragment	7✔
87
88	@property	7✔
89	def help_text(self) -> str:	7✔
NEW 90	return self.schema.get("description", "")	×
91
92	def __repr__(self) -> str:
93	args = [repr(self.tool), self.id]
94	if self.fragment:
95	args.append(f"fragment={self.fragment!r}")
96	return f"{self.__class__.__name__}({', '.join(args)}, <schema: {self.id}>)"
97
98
99	if typing.TYPE_CHECKING:
100	_: PluginProtocol = typing.cast(PluginWrapper, None)
101
102
103	def iterate_entry_points(group: str) -> Iterable[EntryPoint]:	7✔
104	"""Produces a generator yielding an EntryPoint object for each plugin registered
105	via ``setuptools`` `entry point`_ mechanism.
106
107	This method can be used in conjunction with :obj:`load_from_entry_point` to filter
108	the plugins before actually loading them. The entry points are not
109	deduplicated, but they are sorted.
110	"""
111	entries = entry_points()	7✔
112	if hasattr(entries, "select"): # pragma: no cover
113	# The select method was introduced in importlib_metadata 3.9 (and Python 3.10)
114	# and the previous dict interface was declared deprecated
115	select = typing.cast(
116	Any,
117	getattr(entries, "select"), # noqa: B009
118	) # typecheck gymnastics
119	entries_: Iterable[EntryPoint] = select(group=group)
120	else: # pragma: no cover
121	# TODO: Once Python 3.10 becomes the oldest version supported, this fallback and
122	# conditional statement can be removed.
123	entries_ = (plugin for plugin in entries.get(group, []))
124	return sorted(entries_, key=lambda e: e.name)	7✔
125
126
127	def load_from_entry_point(entry_point: EntryPoint) -> PluginWrapper:	7✔
128	"""Carefully load the plugin, raising a meaningful message in case of errors"""
129	try:	7✔
130	fn = entry_point.load()	7✔
131	return PluginWrapper(entry_point.name, fn)	7✔
132	except Exception as ex:	7✔
133	raise ErrorLoadingPlugin(entry_point=entry_point) from ex	7✔
134
135
136	def load_from_multi_entry_point(	7✔
137	entry_point: EntryPoint,
138	) -> Generator[StoredPlugin, None, None]:
139	"""Carefully load the plugin, raising a meaningful message in case of errors"""
140	try:	7✔
141	fn = entry_point.load()	7✔
142	output = fn()	7✔
NEW UNCOV 143	except Exception as ex:	×
NEW UNCOV 144	raise ErrorLoadingPlugin(entry_point=entry_point) from ex	×
145
146	for tool, schema in output.get("tools", {}).items():	7✔
147	yield StoredPlugin(tool, schema)	7✔
148	for schema in output.get("schemas", []):	7✔
149	yield StoredPlugin("", schema)	7✔
150
151
152	def list_from_entry_points(	7✔
153	filtering: Callable[[EntryPoint], bool] = lambda _: True,
154	) -> List[Union[PluginWrapper, StoredPlugin]]:
155	"""Produces a list of plugin objects for each plugin registered
156	via ``setuptools`` `entry point`_ mechanism.
157
158	Args:
159	filtering: function returning a boolean deciding if the entry point should be
160	loaded and included (or not) in the final list. A ``True`` return means the
161	plugin should be included.
162	"""
163	eps: List[Union[PluginWrapper, StoredPlugin]] = [	7✔
164	load_from_entry_point(e)
165	for e in iterate_entry_points("validate_pyproject.tool_schema")
166	if filtering(e)
167	]
168	for e in iterate_entry_points("validate_pyproject.multi_schema"):	7✔
169	eps.extend(load_from_multi_entry_point(e))	7✔
170	dedup = {(e.tool if e.tool else e.id): e for e in sorted(eps, key=lambda e: e.tool)}	7✔
171	return list(dedup.values())	7✔
172
173
174	class ErrorLoadingPlugin(RuntimeError):	7✔
175	_DESC = """There was an error loading '{plugin}'.	7✔
176	Please make sure you have installed a version of the plugin that is compatible
177	with {package} {version}. You can also try uninstalling it.
178	"""
179	__doc__ = _DESC	7✔
180
181	def __init__(self, plugin: str = "", entry_point: Optional[EntryPoint] = None):	7✔
182	if entry_point and not plugin:	7!
183	plugin = getattr(entry_point, "module", entry_point.name)	7✔
184
185	sub = {"package": __package__, "version": __version__, "plugin": plugin}	7✔
186	msg = dedent(self._DESC).format(**sub).splitlines()	7✔
187	super().__init__(f"{msg[0]}\n{' '.join(msg[1:])}")	7✔

abravalheri / validate-pyproject / 6380044383158272

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