Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/airflow/sdk/_shared/module_loading/__init__.py: 45%

1# 

2# Licensed to the Apache Software Foundation (ASF) under one 

3# or more contributor license agreements. See the NOTICE file 

4# distributed with this work for additional information 

5# regarding copyright ownership. The ASF licenses this file 

6# to you under the Apache License, Version 2.0 (the 

7# "License"); you may not use this file except in compliance 

8# with the License. You may obtain a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, 

13# software distributed under the License is distributed on an 

14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

15# KIND, either express or implied. See the License for the 

16# specific language governing permissions and limitations 

17# under the License. 

18from __future__ import annotations 

19 

20import functools 

21import logging 

22import pkgutil 

23import sys 

24from collections import defaultdict 

25from collections.abc import Callable, Iterator 

26from importlib import import_module 

27from typing import TYPE_CHECKING 

28 

29from .file_discovery import ( 

30 find_path_from_directory as find_path_from_directory, 

31) 

32 

33if sys.version_info >= (3, 12): 

34 from importlib import metadata 

35else: 

36 import importlib_metadata as metadata 

37 

38log = logging.getLogger(__name__) 

39 

40EPnD = tuple[metadata.EntryPoint, metadata.Distribution] 

41 

42if TYPE_CHECKING: 

43 from types import ModuleType 

44 

45 

46def import_string(dotted_path: str): 

47 """ 

48 Import a dotted module path and return the attribute/class designated by the last name in the path. 

49 

50 Raise ImportError if the import failed. 

51 """ 

52 # TODO: Add support for nested classes. Currently, it only works for top-level classes. 

53 try: 

54 module_path, class_name = dotted_path.rsplit(".", 1) 

55 except ValueError: 

56 raise ImportError(f"{dotted_path} doesn't look like a module path") 

57 

58 module = import_module(module_path) 

59 

60 try: 

61 return getattr(module, class_name) 

62 except AttributeError: 

63 raise ImportError(f'Module "{module_path}" does not define a "{class_name}" attribute/class') 

64 

65 

66def qualname(o: object | Callable, use_qualname: bool = False) -> str: 

67 """Convert an attribute/class/callable to a string importable by ``import_string``.""" 

68 if callable(o) and hasattr(o, "__module__"): 

69 if use_qualname and hasattr(o, "__qualname__"): 

70 return f"{o.__module__}.{o.__qualname__}" 

71 if hasattr(o, "__name__"): 

72 return f"{o.__module__}.{o.__name__}" 

73 

74 cls = o 

75 

76 if not isinstance(cls, type): # instance or class 

77 cls = type(cls) 

78 

79 name = cls.__qualname__ 

80 module = cls.__module__ 

81 

82 if module and module != "__builtin__": 

83 return f"{module}.{name}" 

84 

85 return name 

86 

87 

88def iter_namespace(ns: ModuleType): 

89 return pkgutil.iter_modules(ns.__path__, ns.__name__ + ".") 

90 

91 

92def is_valid_dotpath(path: str) -> bool: 

93 """ 

94 Check if a string follows valid dotpath format (ie: 'package.subpackage.module'). 

95 

96 :param path: String to check 

97 """ 

98 import re 

99 

100 if not isinstance(path, str): 

101 return False 

102 

103 # Pattern explanation: 

104 # ^ - Start of string 

105 # [a-zA-Z_] - Must start with letter or underscore 

106 # [a-zA-Z0-9_] - Following chars can be letters, numbers, or underscores 

107 # (\.[a-zA-Z_][a-zA-Z0-9_]*)* - Can be followed by dots and valid identifiers 

108 # $ - End of string 

109 pattern = r"^[a-zA-Z_][a-zA-Z0-9_]*(\.[a-zA-Z_][a-zA-Z0-9_]*)*$" 

110 

111 return bool(re.match(pattern, path)) 

112 

113 

114@functools.cache 

115def _get_grouped_entry_points() -> dict[str, list[EPnD]]: 

116 mapping: dict[str, list[EPnD]] = defaultdict(list) 

117 for dist in metadata.distributions(): 

118 try: 

119 for e in dist.entry_points: 

120 mapping[e.group].append((e, dist)) 

121 except Exception as e: 

122 log.warning("Error when retrieving package metadata (skipping it): %s, %s", dist, e) 

123 return mapping 

124 

125 

126def entry_points_with_dist(group: str) -> Iterator[EPnD]: 

127 """ 

128 Retrieve entry points of the given group. 

129 

130 This is like the ``entry_points()`` function from ``importlib.metadata``, 

131 except it also returns the distribution the entry point was loaded from. 

132 

133 Note that this may return multiple distributions to the same package if they 

134 are loaded from different ``sys.path`` entries. The caller site should 

135 implement appropriate deduplication logic if needed. 

136 

137 :param group: Filter results to only this entrypoint group 

138 :return: Generator of (EntryPoint, Distribution) objects for the specified groups 

139 """ 

140 return iter(_get_grouped_entry_points()[group])