_loader.py 1.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758
  1. from __future__ import annotations
  2. import sys
  3. import warnings
  4. from typing import TYPE_CHECKING, Iterable
  5. from typing_extensions import Final
  6. if sys.version_info >= (3, 8):
  7. import importlib.metadata as importlib_metadata
  8. else:
  9. import importlib_metadata
  10. if TYPE_CHECKING:
  11. from . import PydanticPluginProtocol
  12. PYDANTIC_ENTRY_POINT_GROUP: Final[str] = 'pydantic'
  13. # cache of plugins
  14. _plugins: dict[str, PydanticPluginProtocol] | None = None
  15. # return no plugins while loading plugins to avoid recursion and errors while import plugins
  16. # this means that if plugins use pydantic
  17. _loading_plugins: bool = False
  18. def get_plugins() -> Iterable[PydanticPluginProtocol]:
  19. """Load plugins for Pydantic.
  20. Inspired by: https://github.com/pytest-dev/pluggy/blob/1.3.0/src/pluggy/_manager.py#L376-L402
  21. """
  22. global _plugins, _loading_plugins
  23. if _loading_plugins:
  24. # this happens when plugins themselves use pydantic, we return no plugins
  25. return ()
  26. elif _plugins is None:
  27. _plugins = {}
  28. # set _loading_plugins so any plugins that use pydantic don't themselves use plugins
  29. _loading_plugins = True
  30. try:
  31. for dist in importlib_metadata.distributions():
  32. for entry_point in dist.entry_points:
  33. if entry_point.group != PYDANTIC_ENTRY_POINT_GROUP:
  34. continue
  35. if entry_point.value in _plugins:
  36. continue
  37. try:
  38. _plugins[entry_point.value] = entry_point.load()
  39. except (ImportError, AttributeError) as e:
  40. warnings.warn(
  41. f'{e.__class__.__name__} while loading the `{entry_point.name}` Pydantic plugin, '
  42. f'this plugin will not be installed.\n\n{e!r}'
  43. )
  44. finally:
  45. _loading_plugins = False
  46. return _plugins.values()