feat: multi-plugins with extra schemas #231
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7,15 +7,24 @@ | |
|
|
||
| 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 | ||
|
|
@@ -66,34 +75,63 @@ 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 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: | ||
|
|
@@ -105,23 +143,64 @@ def load_from_entry_point(entry_point: EntryPoint) -> PluginWrapper: | |
| 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["tools"].items(): | ||
| yield StoredPlugin(tool, schema) | ||
| for schema in output.get("schemas", []): | ||
| yield StoredPlugin("", schema) | ||
|
|
||
|
|
||
| class _SortablePlugin(NamedTuple): | ||
| priority: int | ||
| name: str | ||
| plugin: Union[PluginWrapper, StoredPlugin] | ||
|
|
||
| def __lt__(self, other: Any) -> bool: | ||
| return (self.plugin.tool or self.plugin.id, self.name, self.priority) < ( | ||
| other.plugin.tool or other.plugin.id, | ||
| other.name, | ||
| other.priority, | ||
| ) | ||
|
|
||
|
|
||
| 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. | ||
| """ | ||
| 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 sorted( | ||
| iterate_entry_points("validate_pyproject.multi_schema"), | ||
| key=lambda e: e.name, | ||
| reverse=True, | ||
| ) | ||
|
Comment on lines
+193
to
+197
There was a problem hiding this comment.
Thank you very much @henryiii. I am trying to understand the need of sorting twice to ensure replicability. I wonder the following: If we produce the iterators as: tool_eps = (
(0, e.name, load_from_entry_point(e))
...
)
multi_eps = (
((1, e.name, p) for p in load_from_multi_entry_point(e))
...
)Could we remove the def _predictable_sorting(element):
priority, name, plugin = element
return (priority, name, plugin.tool or plugin.id)And then sorting = sorted(eps, key=_predictable_sorting)
dedup = {(e.tool or e.id): e for _, _, e in sorting}(All code is untested) Observation 1: Not sure we need There was a problem hiding this comment. This is not sorting1 twice, and it is not related to which type of entry point takes priority; to switch that, we just chain in the other order. This sort is sorting the multi entry points by name; this happens before they are called and it produces the Schemas. If there are two multi entry points producing 5 tools, this sort is over two items, the later sort is over the five tools. It adds stability if two plugins both define a multi entry point that produces the same name. So, for example, let's say a user has two plugins:
By sorting the multi entry points before rendering them, that ensures that Also, if a user has Footnotes
There was a problem hiding this comment. Ah, I see you are adding the name information. I can try that. There was a problem hiding this comment. Okay, pushed something. |
||
| for p in load_from_multi_entry_point(e) | ||
| if filtering(e) | ||
| ) | ||
| eps = chain(tool_eps, multi_eps) | ||
| dedup = {e.plugin.tool or e.plugin.id: e.plugin for e in sorted(eps, reverse=True)} | ||
| return list(dedup.values())[::-1] | ||
|
|
||
|
|
||
| class ErrorLoadingPlugin(RuntimeError): | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I find it easier to read
Iterator[T]instead ofGenerator[T, None, None]and understand what the function does (we don't have to remember which argument order for the return, yield and send).Since it does not affect too much the typechecking, I tend to stick with the simpler approach. Is there a benefit in using
Generator?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is a habit I have due to a problem in typing: things like
contextlib.contextmanagertakes anIteratorbecause people usually type these as iterators. This is incorrect, ascontextmanagercalls.close, which is part of the Generator protocol, not the Iterator protocol. So mypy can't find a mistake like passing anIteratorinstead of aGeneratorto things likecontextmanager. The reason they gave for not fixing it was that everyone types these asIterator. So I never type it asIterator. :) Python 3.13 also added single argumentGenerator[T], largely to help with this I believe.