Coverage for src/metador_core/schema/plugins.py: 100%

1"""Schemas needed for the plugin system.""" 

2from __future__ import annotations 

3 

4import json 

5from collections import ChainMap 

6from functools import total_ordering 

7from typing import ClassVar, Dict, List, Literal, Optional 

8 

9from pydantic import AnyHttpUrl, Extra, ValidationError, create_model 

10 

11from ..util import is_public_name 

12from .core import BaseModelPlus, MetadataSchema 

13from .types import NonEmptyStr, SemVerTuple, to_semver_str 

14 

15 

16@total_ordering 

17class PluginRef(MetadataSchema): 

18 """Reference to a metador plugin. 

19 

20 This class can be subclassed for specific "marker schemas". 

21 

22 It is not registered as a schema plugin, because it is too general on its own. 

23 """ 

24 

25 class Config: 

26 extra = Extra.forbid 

27 allow_mutation = False 

28 

29 group: NonEmptyStr 

30 """Metador pluggable group name, i.e. name of the entry point group.""" 

31 

32 name: NonEmptyStr 

33 """Registered entry point name inside an entry point group.""" 

34 

35 version: SemVerTuple 

36 """Version of the Python package.""" 

37 

38 def __eq__(self, other): 

39 return ( 

40 self.group == other.group 

41 and self.name == other.name 

42 and self.version == other.version 

43 ) 

44 

45 def __ge__(self, other): 

46 if self.group != other.group: 

47 return self.group >= other.group 

48 if self.name != other.name: 

49 return self.name >= other.name 

50 if self.version != other.version: 

51 return self.version >= other.version 

52 

53 def __hash__(self): 

54 # needed because otherwise would differ in subclass, 

55 # causing problems for equality based on __hash__ 

56 return hash((self.group, self.name, self.version)) 

57 

58 def __str__(self) -> str: 

59 return self.json() # no indent (in contrast to base model) 

60 

61 def supports(self, other: PluginRef) -> bool: 

62 """Return whether this plugin supports objects marked by given reference. 

63 

64 True iff the package name, plugin group and plugin name agree, 

65 and the package of this reference has equal or larger minor version. 

66 """ 

67 if self.group != other.group: 

68 return False 

69 if self.name != other.name: 

70 return False 

71 if self.version[0] != other.version[0]: # major 

72 return False 

73 if self.version[1] < other.version[1]: # minor 

74 return False 

75 return True 

76 

77 @classmethod 

78 def _subclass_for(cls, group: NonEmptyStr): 

79 """Create a subclass of PluginRef with group field pre-set.""" 

80 return create_model(f"PG{group.capitalize()}.PluginRef", __base__=cls, group=(Literal[group], group)) # type: ignore 

81 

82 

83class PluginBase(BaseModelPlus): 

84 """All Plugin inner classes must be called `Plugin` and inherit from this class.""" 

85 

86 group: ClassVar[str] = "" # auto-set during plugin group init 

87 

88 # for type checking (mirrors Fields) 

89 name: str 

90 version: SemVerTuple 

91 requires: List[PluginRef] = [] 

92 

93 def ref(self, *, version: Optional[SemVerTuple] = None): 

94 from ..plugins import plugingroups 

95 

96 assert self.group, "Must be called from a subclass that has group set!" 

97 return plugingroups[self.group].PluginRef( 

98 name=self.name, version=version or self.version 

99 ) 

100 

101 def plugin_string(self): 

102 return f"metador.{self.group}.{self.name}.{to_semver_str(self.version)}" 

103 

104 def __str__(self) -> str: 

105 # pretty-print semver in user-facing representation 

106 dct = dict( 

107 group=self.group, name=self.name, version=to_semver_str(self.version) 

108 ) 

109 dct.update( 

110 self.json_dict(exclude_defaults=True, exclude={"name", "group", "version"}) 

111 ) 

112 return json.dumps(dct, indent=2) 

113 

114 @classmethod 

115 def parse_info(cls, info, *, ep_name: str = ""): 

116 if isinstance(info, cls): 

117 return info # nothing to do, already converted info class to PluginBase (sub)model 

118 

119 expected_ep_name = f"{info.name}__{to_semver_str(info.version)}" 

120 ep_name = ep_name or expected_ep_name 

121 if ep_name != expected_ep_name: 

122 msg = f"{ep_name}: Based on plugin info, entrypoint must be called '{expected_ep_name}'!" 

123 raise ValueError(msg) 

124 try: 

125 fields = ChainMap( 

126 *(c.__dict__ for c in info.__mro__) 

127 ) # this will treat inheritance well 

128 # validate 

129 public_attrs = {k: v for k, v in fields.items() if is_public_name(k)} 

130 return cls(**public_attrs) 

131 except ValidationError as e: 

132 raise TypeError(f"{ep_name}: {ep_name}.Plugin validation error: \n{str(e)}") 

133 

134 

135PkgPlugins = Dict[str, List[PluginRef]] 

136"""Dict from plugin group name to plugins provided by a package.""" 

137 

138 

139class PluginPkgMeta(MetadataSchema): 

140 """Metadata of a Python package containing Metador plugins.""" 

141 

142 name: NonEmptyStr 

143 """Name of the Python package.""" 

144 

145 version: SemVerTuple 

146 """Version of the Python package.""" 

147 

148 repository_url: Optional[AnyHttpUrl] = None 

149 """Python package source location (pip-installable / git-clonable).""" 

150 

151 plugins: PkgPlugins = {} 

152 

153 @classmethod 

154 def for_package(cls, package_name: str) -> PluginPkgMeta: 

155 """Extract metadata about a Metador plugin providing Python package.""" 

156 # avoid circular import by importing here 

157 from importlib_metadata import distribution 

158 

159 from ..plugin.entrypoints import DistMeta, distmeta_for 

160 from ..plugin.types import EPName, from_ep_name 

161 

162 dm: DistMeta = distmeta_for(distribution(package_name)) 

163 

164 plugins: PkgPlugins = {} 

165 for group, ep_names in dm.plugins.items(): 

166 plugins[group] = [] 

167 for ep_name in ep_names: 

168 name, version = from_ep_name(EPName(ep_name)) 

169 ref = PluginRef(group=group, name=name, version=version) 

170 plugins[group].append(ref) 

171 

172 return cls( 

173 name=dm.name, 

174 version=dm.version, 

175 repository_url=dm.repository_url, 

176 plugins=plugins, 

177 )