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

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 inspect 

22import logging 

23import pkgutil 

24import re 

25import sys 

26from collections import defaultdict 

27from collections.abc import Callable, Iterator 

28from importlib import import_module 

29from typing import TYPE_CHECKING 

30 

31from .dag_file import ( 

32 MODIFIED_DAG_MODULE_NAME as MODIFIED_DAG_MODULE_NAME, 

33 UNUSUAL_MODULE_PREFIX as UNUSUAL_MODULE_PREFIX, 

34 accepts_dag_definition as accepts_dag_definition, 

35 get_unique_dag_module_name as get_unique_dag_module_name, 

36 might_contain_dag as might_contain_dag, 

37 might_contain_dag_via_default_heuristic as might_contain_dag_via_default_heuristic, 

38) 

39from .file_discovery import ( 

40 find_path_from_directory as find_path_from_directory, 

41) 

42 

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

44 from importlib import metadata 

45else: 

46 import importlib_metadata as metadata 

47 

48log = logging.getLogger(__name__) 

49 

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

51 

52if TYPE_CHECKING: 

53 from types import ModuleType 

54 

55 

56def accepts_context(callback: Callable) -> bool: 

57 """Check if callback accepts a 'context' parameter or **kwargs.""" 

58 try: 

59 sig = inspect.signature(callback) 

60 except (ValueError, TypeError): 

61 return True 

62 params = sig.parameters 

63 return "context" in params or any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values()) 

64 

65 

66def accepts_keyword_args(func: Callable) -> bool: 

67 """Check if a callable accepts any keyword arguments (named params or **kwargs).""" 

68 try: 

69 sig = inspect.signature(func) 

70 except (ValueError, TypeError): 

71 return True 

72 return any( 

73 p.kind 

74 in ( 

75 inspect.Parameter.POSITIONAL_OR_KEYWORD, 

76 inspect.Parameter.KEYWORD_ONLY, 

77 inspect.Parameter.VAR_KEYWORD, 

78 ) 

79 for p in sig.parameters.values() 

80 ) 

81 

82 

83def import_string(dotted_path: str): 

84 """ 

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

86 

87 Raise ImportError if the import failed. 

88 """ 

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

90 try: 

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

92 except ValueError: 

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

94 

95 module = import_module(module_path) 

96 

97 try: 

98 return getattr(module, class_name) 

99 except AttributeError: 

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

101 

102 

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

104 """ 

105 Convert an attribute/class/callable to a string. 

106 

107 By default, returns a string importable by ``import_string`` (includes module path). 

108 With exclude_module=True, returns only the qualified name without module prefix, 

109 useful for stable identification across deployments where module paths may vary. 

110 """ 

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

112 if exclude_module: 

113 if hasattr(o, "__qualname__"): 

114 return o.__qualname__ 

115 if hasattr(o, "__name__"): 

116 return o.__name__ 

117 # Handle functools.partial objects specifically (not just any object with 'func' attr) 

118 if isinstance(o, functools.partial): 

119 return qualname(o.func, exclude_module=True) 

120 return type(o).__qualname__ 

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

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

123 if hasattr(o, "__name__"): 

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

125 

126 cls = o 

127 

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

129 cls = type(cls) 

130 

131 name = cls.__qualname__ 

132 module = cls.__module__ 

133 

134 if exclude_module: 

135 return name 

136 

137 if module and module != "__builtin__": 

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

139 

140 return name 

141 

142 

143def iter_namespace(ns: ModuleType): 

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

145 

146 

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

148 """ 

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

150 

151 :param path: String to check 

152 """ 

153 if not isinstance(path, str): 

154 return False 

155 

156 # Pattern explanation: 

157 # ^ - Start of string 

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

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

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

161 # $ - End of string 

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

163 

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

165 

166 

167def _normalize_dist_name(name: str) -> str: 

168 # PEP 503 normalization, the same key importlib.metadata.entry_points() deduplicates on. 

169 return re.sub(r"[-_.]+", "-", name).lower() 

170 

171 

172@functools.cache 

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

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

175 seen_dist_names: set[str] = set() 

176 for dist in metadata.distributions(): 

177 try: 

178 # A distribution is yielded once per sys.path entry it is found on (e.g. RHEL/Fedora 

179 # venvs list both lib64/ and lib/ site-packages), possibly in different versions. 

180 # Only the first one is importable, so it is the only one whose entry points count. 

181 dist_metadata = dist.metadata 

182 if dist_metadata and (name := dist_metadata.get("Name")): 

183 normalized_name = _normalize_dist_name(name) 

184 if normalized_name in seen_dist_names: 

185 continue 

186 seen_dist_names.add(normalized_name) 

187 for e in dist.entry_points: 

188 # importlib_metadata >= 10 types this as its IDistribution protocol, which older 

189 # releases and the stdlib lack; a cast would be redundant on those. 

190 mapping[e.group].append((e, dist)) # type: ignore[arg-type] 

191 except Exception as e: 

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

193 return mapping 

194 

195 

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

197 """ 

198 Retrieve entry points of the given group. 

199 

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

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

202 

203 Like ``entry_points()``, a distribution found on several ``sys.path`` entries 

204 is only returned once: the first one found, which is the one that is imported. 

205 

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

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

208 """ 

209 return iter(_get_grouped_entry_points()[group])