Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/ge/ge/_internal/plugin_loader.py: 92%

1#!/usr/bin/env python3 

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

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

4# Copyright (c) 2026 Huawei Technologies Co., Ltd. 

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

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

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

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

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

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

11# ----------------------------------------------------------------------------------------------------------- 

12 

13"""Shared Python plugin loading helpers.""" 

14 

15from __future__ import annotations 

16 

17import importlib 

18import importlib.util 

19import os 

20import sys 

21from pathlib import Path 

22from types import ModuleType 

23from typing import Dict, Iterable, List, Optional 

24 

25 

26_MODULE_NAME_BY_CANONICAL_PATH: Dict[Path, str] = {} 

27 

28 

29def normalize_path_list(path_value: str) -> List[str]: 

30 if not path_value: 

31 return [] 

32 return [item.strip() for item in path_value.split(os.pathsep) if item.strip()] 

33 

34 

35def _get_loaded_module(canonical_path: Path) -> Optional[ModuleType]: 

36 module_name = _MODULE_NAME_BY_CANONICAL_PATH.get(canonical_path) 

37 if module_name is None: 

38 return None 

39 module = sys.modules.get(module_name) 

40 if module is None: 

41 del _MODULE_NAME_BY_CANONICAL_PATH[canonical_path] 

42 return module 

43 

44 

45def load_module_from_file( 

46 file_path: Path, *, module_prefix: str, plugin_kind: str 

47) -> ModuleType: 

48 file_path = file_path.resolve() 

49 loaded_module = _get_loaded_module(file_path) 

50 if loaded_module is not None: 

51 return loaded_module 

52 module_name = f"{module_prefix}{file_path.stem}_{abs(hash(str(file_path)))}" 

53 if module_name in sys.modules: 

54 return sys.modules[module_name] 

55 spec = importlib.util.spec_from_file_location(module_name, file_path) 

56 if spec is None or spec.loader is None: 

57 raise ImportError(f"cannot load {plugin_kind} file: {file_path}") 

58 module = importlib.util.module_from_spec(spec) 

59 sys.modules[module_name] = module 

60 try: 

61 spec.loader.exec_module(module) 

62 except BaseException: 

63 del sys.modules[module_name] 

64 raise 

65 _MODULE_NAME_BY_CANONICAL_PATH[file_path] = module_name 

66 return module 

67 

68 

69def scan_modules_from_path( 

70 path_item: str, *, module_prefix: str, plugin_kind: str 

71) -> List[ModuleType]: 

72 path = Path(path_item).resolve() 

73 if not path.exists(): 

74 raise FileNotFoundError(f"{plugin_kind} path does not exist: {path}") 

75 if path.is_file(): 

76 if path.suffix != ".py": 

77 raise ValueError(f"{plugin_kind} file must end with .py: {path}") 

78 return [ 

79 load_module_from_file( 

80 path, module_prefix=module_prefix, plugin_kind=plugin_kind 

81 ) 

82 ] 

83 

84 modules: List[ModuleType] = [] 

85 path_str = str(path) 

86 path_inserted = False 

87 if path_str not in sys.path: 

88 sys.path.insert(0, path_str) 

89 path_inserted = True 

90 try: 

91 for child in sorted(path.iterdir(), key=lambda item: item.name): 

92 if child.name.startswith("_"): 

93 continue 

94 if child.is_file() and child.suffix == ".py": 

95 modules.append( 

96 load_module_from_file( 

97 child, module_prefix=module_prefix, plugin_kind=plugin_kind 

98 ) 

99 ) 

100 continue 

101 if child.is_dir() and (child / "__init__.py").exists(): 

102 package_init = (child / "__init__.py").resolve() 

103 module = _get_loaded_module(package_init) 

104 if module is None: 

105 module = importlib.import_module(child.name) 

106 _MODULE_NAME_BY_CANONICAL_PATH[package_init] = module.__name__ 

107 modules.append(module) 

108 finally: 

109 if path_inserted: 

110 sys.path.remove(path_str) 

111 return modules 

112 

113 

114def load_plugins_from_env( 

115 env_name: str, *, module_prefix: str, plugin_kind: str 

116) -> List[ModuleType]: 

117 loaded_modules: List[ModuleType] = [] 

118 seen_module_names = set() 

119 

120 def append_modules(modules: Iterable[ModuleType]) -> None: 

121 for module in modules: 

122 module_name = getattr(module, "__name__", "") 

123 if module_name in seen_module_names: 

124 continue 

125 seen_module_names.add(module_name) 

126 loaded_modules.append(module) 

127 

128 for path_item in normalize_path_list(os.getenv(env_name, "")): 

129 append_modules( 

130 scan_modules_from_path( 

131 path_item, module_prefix=module_prefix, plugin_kind=plugin_kind 

132 ) 

133 ) 

134 return loaded_modules