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