Coverage for klayout_pex/plugin/registry.py: 95%

1# 

2# -------------------------------------------------------------------------------- 

3# SPDX-FileCopyrightText: 2024-2026 Martin Jan Köhler and Harald Pretl 

4# Johannes Kepler University, Institute for Integrated Circuits. 

5# 

6# This file is part of KPEX 

7# (see https://github.com/iic-jku/klayout-pex). 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <http://www.gnu.org/licenses/>. 

21# SPDX-License-Identifier: GPL-3.0-or-later 

22# -------------------------------------------------------------------------------- 

23# 

24 

25"""Lazy discovery and selection of plugins from one entry-point group.""" 

26 

27from __future__ import annotations 

28 

29from dataclasses import dataclass 

30from importlib import metadata 

31import inspect 

32import re 

33from typing import Any, Callable, ClassVar, List, Mapping, Optional, Tuple, Type 

34 

35from ..version import __version__ 

36 

37 

38@dataclass(frozen=True) 

39class PluginInfo: 

40 """Metadata available without importing the plugin implementation.""" 

41 

42 name: str 

43 distribution: str 

44 version: str = '' 

45 summary: str = '' 

46 """The package summary; for a built-in, the first line of its factory's docstring.""" 

47 

48 target: str = '' 

49 """The ``module:attribute`` the factory is loaded from.""" 

50 

51 @property 

52 def qualified_name(self) -> str: 

53 """An explicit selector, e.g. ``klayout-pex:fastercap``. 

54 

55 Distribution names follow PyPA normalization. Plugin names remain 

56 case-sensitive. This requires no external packaging dependency. 

57 """ 

58 distribution = re.sub(r'[-_.]+', '-', self.distribution).lower() 

59 return f'{distribution}:{self.name}' 

60 

61 

62@dataclass(frozen=True) 

63class PluginRegistration: 

64 """Associate metadata with a loader that defers importing its factory.""" 

65 

66 info: PluginInfo 

67 load_factory: Callable[[], Callable[[], Any]] 

68 builtin: bool = False 

69 

70 

71class PluginRegistry: 

72 """Snapshot one entry-point group and optional built-in plugin factories. 

73 

74 Discovery imports no plugin code. Each ``load`` creates a fresh plugin 

75 instance; failures and name conflicts are reported only for the selected 

76 plugin. Construct another registry to refresh installed metadata. 

77 Subclasses name the capability: its entry-point group, required method 

78 and error types. 

79 """ 

80 

81 kind: ClassVar[str] 

82 """What a plugin is called in messages, e.g. ``exporter``.""" 

83 

84 group: ClassVar[str] 

85 method: ClassVar[str] 

86 error: ClassVar[Type[Exception]] 

87 unavailable: ClassVar[Type[Exception]] 

88 

89 def __init__(self, builtins: Optional[Mapping[str, Callable[[], Any]]] = None): 

90 registrations: List[PluginRegistration] = [ 

91 PluginRegistration(PluginInfo(name, 'klayout-pex', __version__, 

92 (inspect.getdoc(factory) or '').split('\n')[0], 

93 f'{factory.__module__}:{factory.__qualname__}'), 

94 lambda factory=factory: factory, builtin=True) 

95 for name, factory in (builtins or {}).items() 

96 ] 

97 for entry_point in metadata.entry_points(group=self.group): 

98 try: 

99 distribution = entry_point.dist.name if entry_point.dist is not None else None 

100 except (KeyError, ValueError): 

101 distribution = None 

102 if not isinstance(distribution, str) or not distribution.strip(): 

103 distribution = 'unknown' 

104 version, summary = _version_and_summary(entry_point.dist) 

105 registrations.append(PluginRegistration( 

106 PluginInfo(entry_point.name, distribution, version, summary, entry_point.value), 

107 entry_point.load)) 

108 self._registrations: Tuple[PluginRegistration, ...] = tuple(sorted( 

109 registrations, key=lambda registration: registration.info.qualified_name)) 

110 

111 @property 

112 def plugins(self) -> Tuple[PluginInfo, ...]: 

113 """List all registrations, including ambiguous or unavailable ones.""" 

114 return tuple(registration.info for registration in self._registrations) 

115 

116 def selector(self, info: PluginInfo) -> str: 

117 """The shortest selector that loads ``info``: its bare name unless that is taken.""" 

118 try: 

119 if self.get_info(info.name) == info: 

120 return info.name 

121 except self.error: 

122 pass 

123 return info.qualified_name 

124 

125 def get_info(self, selector: str) -> PluginInfo: 

126 """Resolve a selector to its provider metadata without loading code.""" 

127 return self._registration(selector).info 

128 

129 def _registration(self, selector: str) -> PluginRegistration: 

130 if ':' in selector: 

131 distribution, name = selector.split(':', 1) 

132 qualified_name = PluginInfo(name, distribution).qualified_name 

133 matches = [r for r in self._registrations if r.info.qualified_name == qualified_name] 

134 else: 

135 matches = [r for r in self._registrations if r.info.name == selector] 

136 builtins = [r for r in matches if r.builtin] 

137 if builtins: 

138 matches = builtins 

139 

140 if not matches: 

141 choices = ", ".join(info.qualified_name for info in self.plugins) or "(none)" 

142 raise self.error(f"No {self.kind} for '{selector}'. Available {self.kind}s: {choices}") 

143 if len(matches) > 1: 

144 choices = ', '.join(r.info.qualified_name for r in matches) 

145 raise self.error( 

146 f"Ambiguous {self.kind} '{selector}': {choices}. " 

147 "Select a distribution:name; duplicate registrations within one " 

148 "distribution must be removed.") 

149 

150 return matches[0] 

151 

152 def load(self, selector: str) -> Any: 

153 """Load a name or ``distribution:name``; built-ins win for bare names.""" 

154 registration = self._registration(selector) 

155 try: 

156 factory = registration.load_factory() 

157 if not callable(factory): 

158 raise TypeError(f'entry point must be a zero-argument {self.kind} factory') 

159 plugin = factory() 

160 if inspect.isclass(plugin) or not callable(getattr(plugin, self.method, None)): 

161 raise TypeError(f'factory must return an object with an {self.method} method') 

162 except self.error: 

163 raise 

164 except Exception as exc: 

165 raise self.unavailable( 

166 f"Could not load {self.kind} '{registration.info.qualified_name}': " 

167 f"{type(exc).__name__}: {exc}") from exc 

168 return plugin 

169 

170 

171def _version_and_summary(distribution: Optional[metadata.Distribution]) -> Tuple[str, str]: 

172 """Read from package metadata; broken metadata must not hide the plugin.""" 

173 try: 

174 package = distribution.metadata if distribution is not None else None 

175 if package is None: 

176 return '', '' 

177 return package.get('Version') or '', package.get('Summary') or '' 

178 except Exception: 

179 return '', '' 

180 

181 

182__all__ = ['PluginInfo', 'PluginRegistration', 'PluginRegistry']