Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/airflow/_shared/providers_discovery/providers_discovery.py: 39%

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. 

18"""Shared provider discovery utilities.""" 

19 

20from __future__ import annotations 

21 

22import contextlib 

23import json 

24import logging 

25import pathlib 

26from collections.abc import Callable, MutableMapping 

27from dataclasses import dataclass 

28from functools import wraps 

29from importlib.metadata import Distribution 

30from importlib.resources import files as resource_files 

31from time import perf_counter 

32from typing import NamedTuple, ParamSpec, Protocol, cast 

33 

34from packaging.utils import canonicalize_name 

35 

36from ..module_loading import entry_points_with_dist 

37 

38log = logging.getLogger(__name__) 

39 

40 

41PS = ParamSpec("PS") 

42 

43 

44KNOWN_UNHANDLED_OPTIONAL_FEATURE_ERRORS = [("apache-airflow-providers-google", "No module named 'paramiko'")] 

45 

46 

47class ProvidersManagerProtocol(Protocol): 

48 """Protocol for ProvidersManager for type checking purposes.""" 

49 

50 _initialized_cache: dict[str, bool] 

51 

52 

53@dataclass 

54class ProviderInfo: 

55 """ 

56 Provider information. 

57 

58 :param version: version string 

59 :param data: dictionary with information about the provider 

60 """ 

61 

62 version: str 

63 data: dict 

64 

65 

66class HookClassProvider(NamedTuple): 

67 """Hook class and Provider it comes from.""" 

68 

69 hook_class_name: str 

70 package_name: str 

71 

72 

73class HookInfo(NamedTuple): 

74 """Hook information.""" 

75 

76 hook_class_name: str 

77 connection_id_attribute_name: str 

78 package_name: str 

79 hook_name: str 

80 connection_type: str 

81 connection_testable: bool 

82 dialects: list[str] = [] 

83 

84 

85class PluginInfo(NamedTuple): 

86 """Plugin class, name and provider it comes from.""" 

87 

88 name: str 

89 plugin_class: str 

90 provider_name: str 

91 

92 

93class LazyDictWithCache(MutableMapping): 

94 """ 

95 Lazy-loaded cached dictionary. 

96 

97 Dictionary, which in case you set callable, executes the passed callable with `key` attribute 

98 at first use - and returns and caches the result. 

99 """ 

100 

101 __slots__ = ["_resolved", "_raw_dict"] 

102 

103 def __init__(self, *args, **kw): 

104 self._resolved = set() 

105 self._raw_dict = dict(*args, **kw) 

106 

107 def __setitem__(self, key, value): 

108 self._raw_dict.__setitem__(key, value) 

109 

110 def __getitem__(self, key): 

111 value = self._raw_dict.__getitem__(key) 

112 if key not in self._resolved and callable(value): 

113 # exchange callable with result of calling it -- but only once! allow resolver to return a 

114 # callable itself 

115 value = value() 

116 self._resolved.add(key) 

117 self._raw_dict.__setitem__(key, value) 

118 return value 

119 

120 def __delitem__(self, key): 

121 with contextlib.suppress(KeyError): 

122 self._resolved.remove(key) 

123 self._raw_dict.__delitem__(key) 

124 

125 def __iter__(self): 

126 return iter(self._raw_dict) 

127 

128 def __len__(self): 

129 return len(self._raw_dict) 

130 

131 def __contains__(self, key): 

132 return key in self._raw_dict 

133 

134 def clear(self): 

135 self._resolved.clear() 

136 self._raw_dict.clear() 

137 

138 

139def _read_schema_from_resources_or_local_file(filename: str) -> dict: 

140 """Read JSON schema from resources or local file.""" 

141 try: 

142 with resource_files("airflow").joinpath(filename).open("rb") as f: 

143 schema = json.load(f) 

144 except (TypeError, FileNotFoundError): 

145 with (pathlib.Path(__file__).parent / filename).open("rb") as f: 

146 schema = json.load(f) 

147 return schema 

148 

149 

150def _create_provider_info_schema_validator(): 

151 """Create JSON schema validator from the provider_info.schema.json.""" 

152 import jsonschema 

153 

154 schema = _read_schema_from_resources_or_local_file("provider_info.schema.json") 

155 cls = jsonschema.validators.validator_for(schema) 

156 validator = cls(schema) 

157 return validator 

158 

159 

160def _check_builtin_provider_prefix(provider_package: str, class_name: str) -> bool: 

161 """Check if builtin provider class has correct prefix.""" 

162 if provider_package.startswith("apache-airflow"): 

163 provider_path = provider_package[len("apache-") :].replace("-", ".") 

164 if not class_name.startswith(provider_path): 

165 log.warning( 

166 "Coherence check failed when importing '%s' from '%s' package. It should start with '%s'", 

167 class_name, 

168 provider_package, 

169 provider_path, 

170 ) 

171 return False 

172 return True 

173 

174 

175def log_optional_feature_disabled(class_name, e, provider_package): 

176 """Log optional feature disabled.""" 

177 log.debug( 

178 "Optional feature disabled on exception when importing '%s' from '%s' package", 

179 class_name, 

180 provider_package, 

181 exc_info=e, 

182 ) 

183 log.info( 

184 "Optional provider feature disabled when importing '%s' from '%s' package", 

185 class_name, 

186 provider_package, 

187 ) 

188 

189 

190def log_import_warning(class_name, e, provider_package): 

191 """Log import warning.""" 

192 log.warning( 

193 "Exception when importing '%s' from '%s' package", 

194 class_name, 

195 provider_package, 

196 exc_info=e, 

197 ) 

198 

199 

200def provider_info_cache(cache_name: str) -> Callable[[Callable[PS, None]], Callable[PS, None]]: 

201 """ 

202 Decorate and cache provider info. 

203 

204 Decorator factory that create decorator that caches initialization of provider's parameters 

205 :param cache_name: Name of the cache 

206 """ 

207 

208 def provider_info_cache_decorator(func: Callable[PS, None]) -> Callable[PS, None]: 

209 @wraps(func) 

210 def wrapped_function(*args: PS.args, **kwargs: PS.kwargs) -> None: 

211 instance = cast("ProvidersManagerProtocol", args[0]) 

212 

213 if cache_name in instance._initialized_cache: 

214 return 

215 start_time = perf_counter() 

216 log.debug("Initializing Provider Manager[%s]", cache_name) 

217 func(*args, **kwargs) 

218 instance._initialized_cache[cache_name] = True 

219 log.debug( 

220 "Initialization of Provider Manager[%s] took %.2f seconds", 

221 cache_name, 

222 perf_counter() - start_time, 

223 ) 

224 

225 return wrapped_function 

226 

227 return provider_info_cache_decorator 

228 

229 

230def discover_all_providers_from_packages( 

231 provider_dict: dict[str, ProviderInfo], 

232 provider_schema_validator, 

233 incompatibility_reason: Callable[[Distribution], str | None] | None = None, 

234) -> None: 

235 """ 

236 Discover all providers by scanning packages installed. 

237 

238 The list of providers should be returned via the 'apache_airflow_provider' 

239 entrypoint as a dictionary conforming to the 'airflow/provider_info.schema.json' 

240 schema. Note that the schema is different at runtime than provider.yaml.schema.json. 

241 The development version of provider schema is more strict and changes together with 

242 the code. The runtime version is more relaxed (allows for additional properties) 

243 and verifies only the subset of fields that are needed at runtime. 

244 

245 :param provider_dict: Dictionary to populate with discovered providers 

246 :param provider_schema_validator: JSON schema validator for provider info 

247 """ 

248 import jsonschema 

249 

250 for entry_point, dist in entry_points_with_dist("apache_airflow_provider"): 

251 if not dist.metadata: 

252 continue 

253 package_name = canonicalize_name(dist.metadata["name"]) 

254 if package_name in provider_dict: 

255 continue 

256 log.debug("Loading %s from package %s", entry_point, package_name) 

257 version = dist.version 

258 if incompatibility_reason and (reason := incompatibility_reason(dist)): 

259 log.warning(reason) 

260 continue 

261 provider_info = entry_point.load()() 

262 try: 

263 provider_schema_validator.validate(provider_info) 

264 except jsonschema.ValidationError: 

265 # One provider's malformed metadata must not stop the others from loading. This 

266 # used to escape discovery entirely, so every provider after it was skipped and 

267 # initialize_providers_list() raised on this and every later call. 

268 # Only ValidationError is caught: a SchemaError or UnknownType means the schema 

269 # itself is broken, which must stay loud rather than be reported once per provider. 

270 log.warning( 

271 "The provider %s does not conform to the provider_info schema and has been skipped.", 

272 package_name, 

273 exc_info=True, 

274 ) 

275 continue 

276 # 'package-name' is not in the runtime schema's required list, so validation passing 

277 # does not guarantee it is present. 

278 provider_info_package_name = provider_info.get("package-name") 

279 if not provider_info_package_name: 

280 log.warning( 

281 "The provider %s does not declare 'package-name' in its provider info and has been skipped.", 

282 package_name, 

283 ) 

284 continue 

285 if package_name != provider_info_package_name: 

286 raise ValueError( 

287 f"The package '{package_name}' from packaging information " 

288 f"{provider_info_package_name} do not match. Please make sure they are aligned" 

289 ) 

290 

291 # issue-59576: Retrieve the project.urls.documentation from dist.metadata 

292 project_urls = dist.metadata.get_all("Project-URL") 

293 documentation_url: str | None = None 

294 

295 if project_urls: 

296 for entry in project_urls: 

297 if "," in entry: 

298 name, url = entry.split(",", 1) 

299 if name.strip().lower() == "documentation": 

300 documentation_url = url.strip() 

301 break 

302 

303 provider_info["documentation-url"] = documentation_url 

304 

305 if package_name not in provider_dict: 

306 provider_dict[package_name] = ProviderInfo(version, provider_info) 

307 else: 

308 log.warning( 

309 "The provider for package '%s' could not be registered from because providers for that " 

310 "package name have already been registered", 

311 package_name, 

312 )