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

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 Iterable, List 

24 

25 

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

27 if not path_value: 

28 return [] 

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

30 

31 

32def load_module_from_file( 

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

34) -> ModuleType: 

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

36 if module_name in sys.modules: 

37 return sys.modules[module_name] 

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

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

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

41 module = importlib.util.module_from_spec(spec) 

42 sys.modules[module_name] = module 

43 spec.loader.exec_module(module) 

44 return module 

45 

46 

47def scan_modules_from_path( 

48 path_item: str, *, module_prefix: str, plugin_kind: str 

49) -> List[ModuleType]: 

50 path = Path(path_item) 

51 if not path.exists(): 

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

53 if path.is_file(): 

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

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

56 return [ 

57 load_module_from_file( 

58 path, module_prefix=module_prefix, plugin_kind=plugin_kind 

59 ) 

60 ] 

61 

62 modules: List[ModuleType] = [] 

63 if str(path) not in sys.path: 

64 sys.path.insert(0, str(path)) 

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

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

67 continue 

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

69 modules.append( 

70 load_module_from_file( 

71 child, module_prefix=module_prefix, plugin_kind=plugin_kind 

72 ) 

73 ) 

74 continue 

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

76 modules.append(importlib.import_module(child.name)) 

77 return modules 

78 

79 

80def load_plugins_from_env( 

81 env_name: str, *, module_prefix: str, plugin_kind: str 

82) -> List[ModuleType]: 

83 loaded_modules: List[ModuleType] = [] 

84 seen_module_names = set() 

85 

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

87 for module in modules: 

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

89 if module_name in seen_module_names: 

90 continue 

91 seen_module_names.add(module_name) 

92 loaded_modules.append(module) 

93 

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

95 append_modules( 

96 scan_modules_from_path( 

97 path_item, module_prefix=module_prefix, plugin_kind=plugin_kind 

98 ) 

99 ) 

100 return loaded_modules