Coverage for src/metador_core/plugin/util.py: 100%

1"""General utitilies with relevance for plugins.""" 

2from typing import TYPE_CHECKING, Any, Optional, Type, TypeVar 

3 

4from ..util import eprint 

5from .types import PluginLike, is_pluginlike, to_ep_name 

6 

7if TYPE_CHECKING: # pragma: no cover 

8 from .interface import PluginGroup 

9else: 

10 PluginGroup = Any 

11 

12 

13# ---- 

14# helpers for checking plugins (also to be used in PluginGroup subclasses): 

15 

16 

17def implements_method(plugin, base_method): 

18 ep_method = plugin.__dict__.get(base_method.__name__) 

19 return ep_method is not None and base_method != ep_method 

20 

21 

22def check_implements_method(name: str, plugin, base_method): 

23 """Check whether plugin overrides a method of its superclass.""" 

24 if not implements_method(plugin, base_method): 

25 msg = f"{name}: {plugin} does not implement {base_method.__name__}!" 

26 raise TypeError(msg) 

27 

28 

29def check_is_subclass(name: str, plugin, base): 

30 """Check whether plugin has expected parent class (helper method).""" 

31 if not issubclass(plugin, base): 

32 msg = f"{name}: {plugin} is not subclass of {base}!" 

33 raise TypeError(msg) 

34 

35 

36# ---- 

37 

38 

39def is_notebook() -> bool: # pragma: no cover 

40 # https://stackoverflow.com/a/39662359 

41 try: 

42 # get_ipython() is defined globally in ipython-like env! 

43 shell = get_ipython().__class__.__name__ # type: ignore 

44 

45 if shell == "ZMQInteractiveShell": 

46 return True # Jupyter notebook or qtconsole 

47 elif shell == "TerminalInteractiveShell": 

48 return False # Terminal running IPython 

49 else: 

50 return False # Other type (?) 

51 except NameError: 

52 return False # Probably standard Python interpreter 

53 

54 

55T = TypeVar("T", bound=PluginLike) 

56 

57 

58def register_in_group( 

59 pgroup: PluginGroup, 

60 plugin: Optional[Type[T]] = None, 

61 *, 

62 violently: bool = False, 

63): 

64 """Register and load a plugin manually, without defining an entry point.""" 

65 if not violently and not is_notebook(): 

66 raise RuntimeError("This is not supposed to be used outside of notebooks!") 

67 

68 def manual_register(plugin: Type[T]) -> Type[T]: 

69 pginfo = plugin.Plugin 

70 ep_name = to_ep_name(pginfo.name, pginfo.version) 

71 pg_ref = pgroup.PluginRef(name=pginfo.name, version=pginfo.version) 

72 

73 pgroup._ENTRY_POINTS[ep_name] = None 

74 pgroup._LOADED_PLUGINS[pg_ref] = plugin 

75 if pg_ref.name not in pgroup._VERSIONS: 

76 pgroup._VERSIONS[pg_ref.name] = [] 

77 pgroup._VERSIONS[pg_ref.name].append(pg_ref) 

78 

79 pgroup._load_plugin(ep_name, plugin) 

80 if not violently: 

81 eprint( 

82 f"Notebook: Plugin '{pginfo.name}' registered in '{pgroup.name}' group!" 

83 ) # pragma: no cover 

84 return plugin 

85 

86 if not plugin: 

87 return manual_register # used as decorator 

88 else: 

89 if not is_pluginlike(plugin, check_group=False): 

90 raise RuntimeError("This class has no inner Plugin class!") 

91 

92 manual_register(plugin) # used as normal function