Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/ge/ge/es/_plugin_loader.py: 96%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ------------------------------------------------------------------- 

4# ----------------------------------------------------------------------------------------------------------- 

5# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

6# This program is free software, you can redistribute it and/or modify it under the terms and conditions of 

7# CANN Open Software License Agreement Version 2.0 (the "License"). 

8# Please refer to the License for details. You may not use this file except in compliance with the License. 

9# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, 

10# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. 

11# See LICENSE in the root of the software repository for the full text of the License. 

12# ----------------------------------------------------------------------------------------------------------- 

13 

14"""Runtime utilities for dynamically discovering and loading ES plugins.""" 

15 

16import importlib 

17import sys 

18from types import ModuleType 

19from typing import Any, Dict, List 

20 

21try: 

22 from importlib.metadata import entry_points 

23except ImportError: 

24 from importlib_metadata import entry_points # type: ignore 

25 

26_ENTRY_POINT_GROUP = "ge.es.plugins" 

27 

28# Log level keywords 

29LOG_LEVEL_INFO = "INFO" 

30LOG_LEVEL_WARNING = "WARNING" 

31LOG_LEVEL_ERROR = "ERROR" 

32 

33 

34def debug_print(level: str, message: str, *args): 

35 """Print debug message with formatted output. 

36 

37 Args: 

38 level: Log level (INFO, WARNING, ERROR) 

39 message: Message format string 

40 *args: Arguments for message formatting 

41 """ 

42 module_name = __name__ 

43 formatted_message = message % args if args else message 

44 print(f"[{level}] [{module_name}] {formatted_message}") 

45 

46 

47def _iter_plugin_entry_points() -> List[Any]: 

48 """Compatible with entry_points API changes across different Python versions.""" 

49 try: 

50 # Python 3.10+ 

51 candidates = entry_points(group=_ENTRY_POINT_GROUP) 

52 if isinstance(candidates, dict): 

53 return list(candidates.get(_ENTRY_POINT_GROUP, [])) 

54 return list(candidates) 

55 except TypeError: 

56 # Python 3.7-3.9 

57 candidates = entry_points() 

58 if hasattr(candidates, "select"): 

59 return list(candidates.select(group=_ENTRY_POINT_GROUP)) 

60 return list(candidates.get(_ENTRY_POINT_GROUP, [])) 

61 

62 

63def _coerce_to_module(obj: Any, plugin_name: str) -> ModuleType: 

64 """Convert entry point loading result to a module object. 

65 

66 Args: 

67 obj: Object returned by entry point loading 

68 plugin_name: Plugin name for error reporting 

69 

70 Returns: 

71 ModuleType: Converted module object 

72 

73 Raises: 

74 TypeError: If the object type is not supported 

75 """ 

76 if isinstance(obj, ModuleType): 

77 return obj 

78 if isinstance(obj, str): 

79 return importlib.import_module(obj) 

80 if callable(obj): 

81 result = obj() 

82 return _coerce_to_module(result, plugin_name) 

83 raise TypeError( 

84 f"Plugin '{plugin_name}' returned unexpected type: {type(obj).__name__}. " 

85 f"Expected ModuleType, str, or callable returning module." 

86 ) 

87 

88 

89def load_all_plugins() -> Dict[str, ModuleType]: 

90 """Load all plugins registered to ge.es.plugins. 

91 

92 Returns: 

93 dict[str, ModuleType]: Mapping of plugin names to module objects 

94 """ 

95 plugins: Dict[str, ModuleType] = {} 

96 

97 for entry_point in _iter_plugin_entry_points(): 

98 name = getattr(entry_point, "name", None) 

99 if not name: 

100 debug_print( 

101 LOG_LEVEL_WARNING, 

102 "Ignoring ES plugin entry point without name: %s", 

103 entry_point, 

104 ) 

105 continue 

106 

107 try: 

108 # Load entry point 

109 loaded_obj = entry_point.load() 

110 

111 # Convert to module object 

112 module = _coerce_to_module(loaded_obj, name) 

113 

114 # Register to sys.modules to make import ge.es.<name> available 

115 fullname = f"ge.es.{name}" 

116 sys.modules.setdefault(fullname, module) 

117 

118 # Add to plugin dictionary 

119 plugins[name] = module 

120 

121 debug_print(LOG_LEVEL_INFO, "ES plugin '%s' loaded: %s", name, module.__name__) 

122 

123 except AttributeError as err: 

124 debug_print( 

125 LOG_LEVEL_ERROR, 

126 "Failed to load ES plugin '%s': entry point '%s' missing required attribute. " 

127 "Ensure the plugin's __init__.py defines get_module(). Error: %s", 

128 name, 

129 getattr(entry_point, "value", "unknown"), 

130 err, 

131 ) 

132 except ImportError as err: 

133 debug_print( 

134 LOG_LEVEL_ERROR, 

135 "Failed to import ES plugin '%s': %s. Check if all dependencies are installed.", 

136 name, 

137 err, 

138 ) 

139 except Exception as err: 

140 debug_print( 

141 LOG_LEVEL_ERROR, 

142 "Unexpected error loading ES plugin '%s' (entry point: %s): %s", 

143 name, 

144 getattr(entry_point, "value", "unknown"), 

145 err, 

146 ) 

147 

148 return plugins