Coverage for src / dotbot / util / module.py: 93%

1import glob 

2import importlib.util 

3import os 

4from types import ModuleType 

5from typing import List, Optional, Type 

6 

7from dotbot.plugin import Plugin 

8 

9# We keep references to loaded modules so they don't get garbage collected. 

10loaded_modules = [] 

11 

12 

13def load(path: str) -> List[Type[Plugin]]: 

14 basename = os.path.basename(path) 

15 module_name, _ = os.path.splitext(basename) 

16 loaded_module = load_module(module_name, path) 

17 plugins = [] 

18 for name in dir(loaded_module): 

19 possible_plugin = getattr(loaded_module, name) 

20 try: 

21 if issubclass(possible_plugin, Plugin) and possible_plugin is not Plugin: 

22 plugins.append(possible_plugin) 

23 except TypeError: 

24 pass 

25 loaded_modules.append(loaded_module) 

26 return plugins 

27 

28 

29def load_module(module_name: str, path: str) -> ModuleType: 

30 spec = importlib.util.spec_from_file_location(module_name, path) 

31 if not spec or not spec.loader: 

32 msg = f"Unable to load module {module_name} from {path}" 

33 raise ImportError(msg) 

34 module = importlib.util.module_from_spec(spec) 

35 spec.loader.exec_module(module) 

36 return module 

37 

38 

39def load_plugins(paths: List[str], plugins: Optional[List[Type[Plugin]]] = None) -> List[Type[Plugin]]: 

40 """ 

41 Load plugins from the given paths and add them to the given list of plugins. 

42 

43 Args: 

44 paths: List of file paths to load plugins from. Each path can be either a file or a directory. 

45 plugins: List of existing plugins to add to. 

46 

47 Returns the newly-loaded plugins. 

48 """ 

49 if plugins is None: 

50 plugins = [] 

51 new_plugins = [] 

52 plugin_paths = [] 

53 for path in paths: 

54 if os.path.isdir(path): 

55 plugin_paths.extend(glob.glob(os.path.join(path, "*.py"))) 

56 else: 

57 plugin_paths.append(path) 

58 for plugin_path in plugin_paths: 

59 abspath = os.path.abspath(plugin_path) 

60 for plugin in load(abspath): 

61 # ensure plugins are unique to avoid duplicate execution, which 

62 # can happen if, for example, a third-party plugin loads a 

63 # built-in plugin, which will cause it to appear in the list 

64 # returned by load() above 

65 plugin_already_loaded = any( 

66 existing_plugin.__module__ == plugin.__module__ and existing_plugin.__name__ == plugin.__name__ 

67 for existing_plugin in plugins 

68 ) 

69 if not plugin_already_loaded: 

70 plugins.append(plugin) 

71 new_plugins.append(plugin) 

72 return new_plugins