Coverage for src/nendo/main.py: 66%

1# -*- encoding: utf-8 -*- 

2"""Nendo core main class.""" 

3from __future__ import annotations 

4 

5import importlib 

6import inspect 

7import json 

8import logging 

9import os 

10import sys 

11from functools import lru_cache 

12from typing import List, Optional 

13 

14from nendo import library as lib 

15from nendo import schema 

16from nendo.config import NendoConfig, get_settings 

17 

18settings = get_settings() 

19nendo_logger = logging.getLogger("nendo") 

20log_fmt = "[%(asctime)s.%(msecs)03dZ] %(name)s %(levelname)s %(message)s" 

21 

22 

23class Nendo: 

24 """Main class that runs nendo core.""" 

25 

26 _instance = None 

27 library: schema.NendoLibraryPlugin = None 

28 plugins: schema.NendoPluginRegistry = None 

29 logger: logging.Logger = None 

30 config: NendoConfig = None 

31 

32 def __init__( 

33 self, 

34 config: Optional[NendoConfig] = None, 

35 plugins: Optional[List[str]] = None, 

36 logger: Optional[logging.Logger] = None, 

37 ) -> None: 

38 """Create a new nendo instance. 

39 

40 Args: 

41 config (NendoConfig, optional): Custom settings to apply to this instance. 

42 plugins (str, optional): list of plugins to load. Defaults to None. 

43 logger (logging.Logger, optional): Logger to use. Defaults to none. 

44 """ 

45 if hasattr(self, "_initialized") and self._initialized: 

46 return 

47 self.logger = logger or nendo_logger 

48 self.config = config or settings 

49 self.plugins = schema.NendoPluginRegistry() 

50 plugin_names = plugins or self.config.plugins 

51 if self.config.log_file_path == "": 

52 logging.basicConfig( 

53 level=self.config.log_level.upper(), 

54 datefmt="%Y-%m-%dT%H:%M:%S", 

55 format=log_fmt, 

56 ) 

57 else: 

58 if ( 

59 not os.path.isdir(os.path.dirname(self.config.log_file_path)) 

60 or os.path.basename(self.config.log_file_path) == "" 

61 ): 

62 self.logger.error( 

63 "Config var log_file_path has been set but path is not " 

64 "valid or no filename specified. Please fix your configuration.", 

65 ) 

66 sys.exit(-1) 

67 logging.basicConfig( 

68 filename=self.config.log_file_path, 

69 level=self.config.log_level.upper(), 

70 datefmt="%Y-%m-%dT%H:%M:%S", 

71 format=log_fmt, 

72 ) 

73 self._load_plugins(plugin_names=plugin_names) 

74 # initialize nendo library 

75 if self.config.library_plugin == "default": 

76 self.logger.info("Using DuckDBLibrary") 

77 self.library = lib.DuckDBLibrary( 

78 nendo_instance=self, 

79 config=self.config, 

80 logger=self.logger, 

81 plugin_name="DuckDBLibrary", 

82 plugin_version="0.1.1", 

83 ) 

84 else: 

85 self.logger.info(f"Loading {self.config.library_plugin}") 

86 self.library = self._load_plugin(module_name=self.config.library_plugin) 

87 self._initialized = True 

88 

89 # this turns the nendo class effectively into a singleton 

90 def __new__(cls, *args, **kwargs): # noqa: D102, ARG003 

91 if not cls._instance: 

92 cls._instance = super(Nendo, cls).__new__(cls) 

93 cls._instance._initialized = False 

94 return cls._instance 

95 

96 # this makes all library functions callable on the nendo instance itself 

97 def __getattr__(self, attr): 

98 if not hasattr(self.library, attr): 

99 raise AttributeError(f"Function '{attr}' unknown.") 

100 return getattr(self.library, attr) 

101 

102 def _load_plugin(self, module_name: str) -> schema.NendoPlugin: 

103 package_spec = importlib.util.find_spec(module_name) 

104 if package_spec is not None: 

105 try: 

106 module_class = importlib.import_module(module_name) 

107 for cl in inspect.getmembers(module_class, inspect.isclass): 

108 # NOTE currently, we only support one functional class per plugin 

109 if issubclass(cl[1], schema.NendoPlugin): 

110 self.logger.debug( 

111 "Adding Nendo plugin %s from %s", 

112 cl, 

113 module_name, 

114 ) 

115 plugin_name = cl[1].__name__ 

116 plugin_version = importlib.metadata.version(module_name) 

117 plugin_instance: schema.NendoPlugin = getattr( 

118 module_class, 

119 plugin_name, 

120 )( 

121 nendo_instance=self, 

122 config=self.config, 

123 logger=self.logger, 

124 plugin_name=module_name, 

125 plugin_version=plugin_version, 

126 ) 

127 return plugin_instance 

128 except Exception as e: # noqa: BLE001 

129 raise schema.NendoPluginLoadingError( 

130 f"Failed to import plugin '{module_name}'. Error: {e}", 

131 ) from None 

132 else: 

133 raise schema.NendoPluginLoadingError( 

134 f"Plugin {module_name} not installed in system. " 

135 "Please use e.g. pip to install it.", 

136 ) 

137 

138 def _load_plugins(self, plugin_names: List[str] = settings.plugins) -> None: 

139 """Load the specified nendo plugins.""" 

140 for module_name in plugin_names: 

141 plugin_instance = self._load_plugin(module_name=module_name) 

142 self.plugins.add( 

143 plugin_name=plugin_instance.plugin_name, 

144 version=plugin_instance.plugin_version, 

145 plugin_instance=plugin_instance, 

146 ) 

147 

148 plugins_str = str(self.plugins) 

149 for line in plugins_str.split("\n"): 

150 self.logger.info(line) 

151 

152 def __str__(self): 

153 output = "Nendo\n" 

154 output += "Config: " 

155 output += json.dumps(json.loads(self.config.json()), indent=4) 

156 # output += f"Library: {self.library}\n" 

157 output += f"\n{self.plugins}" 

158 return output 

159 

160 

161@lru_cache() 

162def get_nendo(): 

163 """Get the nendo instance singledton, cached.""" 

164 return Nendo()