Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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"""Manages all providers."""
20from __future__ import annotations
22import contextlib
23import functools
24import inspect
25import json
26import logging
27import traceback
28import warnings
29from collections.abc import Callable, Iterator, MutableMapping
30from dataclasses import dataclass
31from functools import wraps
32from importlib.resources import files as resource_files
33from time import perf_counter
34from typing import TYPE_CHECKING, Any, NamedTuple, ParamSpec, TypeVar, cast
36from packaging.utils import canonicalize_name
37from packaging.version import Version
39from airflow import DeprecatedImportWarning
40from airflow._shared.module_loading import import_string
41from airflow._shared.providers_discovery import discover_all_providers_from_packages
42from airflow.exceptions import AirflowOptionalProviderFeatureException, AirflowProviderDeprecationWarning
43from airflow.utils.log.logging_mixin import LoggingMixin
45if TYPE_CHECKING:
46 from importlib.metadata import Distribution
48 from airflow.cli.cli_config import CLICommand
50log = logging.getLogger(__name__)
53PS = ParamSpec("PS")
54RT = TypeVar("RT")
56MIN_PROVIDER_VERSIONS = {
57 "apache-airflow-providers-celery": "2.1.0",
58}
60BLOCKED_PROVIDER_MIN_VERSIONS = {
61 "apache-airflow-providers-common-ai": "1.0.0",
62}
65def _is_source_install(dist: Distribution) -> bool:
66 """Whether the distribution was installed from a local directory, editable or not (PEP 610)."""
67 with contextlib.suppress(ValueError):
68 return "dir_info" in json.loads(dist.read_text("direct_url.json") or "{}")
69 return False
72def provider_incompatibility_reason(dist: Distribution) -> str | None:
73 """Return a reason to skip a provider that cannot run with this Airflow version."""
74 package_name = canonicalize_name(dist.metadata["Name"])
75 minimum_version = BLOCKED_PROVIDER_MIN_VERSIONS.get(package_name)
76 if minimum_version is None or Version(Version(dist.version).base_version) >= Version(minimum_version):
77 return None
78 # A source tree reports the last released version, so version alone can't tell it apart; this
79 # exemption can go as soon as a version > 0.10.0 exists.
80 if _is_source_install(dist):
81 return None
82 return (
83 f"Skipping incompatible provider {package_name} {dist.version}; "
84 f"install {package_name}>={minimum_version} for this version of Airflow."
85 )
88def _ensure_prefix_for_placeholders(field_behaviors: dict[str, Any], conn_type: str):
89 """
90 Verify the correct placeholder prefix.
92 If the given field_behaviors dict contains a placeholder's node, and there
93 are placeholders for extra fields (i.e. anything other than the built-in conn
94 attrs), and if those extra fields are unprefixed, then add the prefix.
96 The reason we need to do this is, all custom conn fields live in the same dictionary,
97 so we need to namespace them with a prefix internally. But for user convenience,
98 and consistency between the `get_ui_field_behaviour` method and the extra dict itself,
99 we allow users to supply the unprefixed name.
100 """
101 conn_attrs = {"host", "schema", "login", "password", "port", "extra"}
103 def ensure_prefix(field):
104 if field not in conn_attrs and not field.startswith("extra__"):
105 return f"extra__{conn_type}__{field}"
106 return field
108 if "placeholders" in field_behaviors:
109 placeholders = field_behaviors["placeholders"]
110 field_behaviors["placeholders"] = {ensure_prefix(k): v for k, v in placeholders.items()}
112 return field_behaviors
115if TYPE_CHECKING:
116 from urllib.parse import SplitResult
118 from airflow.sdk import BaseHook
119 from airflow.sdk.bases.decorator import TaskDecorator
120 from airflow.sdk.definitions.asset import Asset
123class LazyDictWithCache(MutableMapping):
124 """
125 Lazy-loaded cached dictionary.
127 Dictionary, which in case you set callable, executes the passed callable with `key` attribute
128 at first use - and returns and caches the result.
129 """
131 __slots__ = ["_resolved", "_raw_dict"]
133 def __init__(self, *args, **kw):
134 self._resolved = set()
135 self._raw_dict = dict(*args, **kw)
137 def __setitem__(self, key, value):
138 self._raw_dict.__setitem__(key, value)
140 def __getitem__(self, key):
141 value = self._raw_dict.__getitem__(key)
142 if key not in self._resolved and callable(value):
143 # exchange callable with result of calling it -- but only once! allow resolver to return a
144 # callable itself
145 value = value()
146 self._resolved.add(key)
147 self._raw_dict.__setitem__(key, value)
148 return value
150 def __delitem__(self, key):
151 with contextlib.suppress(KeyError):
152 self._resolved.remove(key)
153 self._raw_dict.__delitem__(key)
155 def __iter__(self):
156 return iter(self._raw_dict)
158 def __len__(self):
159 return len(self._raw_dict)
161 def __contains__(self, key):
162 return key in self._raw_dict
164 def clear(self):
165 self._resolved.clear()
166 self._raw_dict.clear()
169def _read_schema_from_resources_or_local_file(filename: str) -> dict:
170 try:
171 with resource_files("airflow").joinpath(filename).open("rb") as f:
172 schema = json.load(f)
173 except (TypeError, FileNotFoundError):
174 import pathlib
176 with (pathlib.Path(__file__).parent / filename).open("rb") as f:
177 schema = json.load(f)
178 return schema
181def _create_provider_info_schema_validator():
182 """Create JSON schema validator from the provider_info.schema.json."""
183 import jsonschema
185 schema = _read_schema_from_resources_or_local_file("provider_info.schema.json")
186 cls = jsonschema.validators.validator_for(schema)
187 validator = cls(schema)
188 return validator
191def _create_customized_form_field_behaviours_schema_validator():
192 """Create JSON schema validator from the customized_form_field_behaviours.schema.json."""
193 import jsonschema
195 schema = _read_schema_from_resources_or_local_file("customized_form_field_behaviours.schema.json")
196 cls = jsonschema.validators.validator_for(schema)
197 validator = cls(schema)
198 return validator
201def _check_builtin_provider_prefix(provider_package: str, class_name: str) -> bool:
202 if provider_package.startswith("apache-airflow"):
203 provider_path = provider_package[len("apache-") :].replace("-", ".")
204 if not class_name.startswith(provider_path):
205 log.warning(
206 "Coherence check failed when importing '%s' from '%s' package. It should start with '%s'",
207 class_name,
208 provider_package,
209 provider_path,
210 )
211 return False
212 return True
215@dataclass
216class ProviderInfo:
217 """
218 Provider information.
220 :param version: version string
221 :param data: dictionary with information about the provider
222 """
224 version: str
225 data: dict
228class HookClassProvider(NamedTuple):
229 """Hook class and Provider it comes from."""
231 hook_class_name: str
232 package_name: str
235class DialectInfo(NamedTuple):
236 """Dialect class and Provider it comes from."""
238 name: str
239 dialect_class_name: str
240 provider_name: str
243class TriggerInfo(NamedTuple):
244 """Trigger class and provider it comes from."""
246 trigger_class_name: str
247 package_name: str
248 integration_name: str
251class NotificationInfo(NamedTuple):
252 """Notification class and provider it comes from."""
254 notification_class_name: str
255 package_name: str
258class RemoteLoggingInfo(NamedTuple):
259 """Remote logging IO handler registered by a provider."""
261 classpath: str
262 scheme: str
263 package_name: str
266class PluginInfo(NamedTuple):
267 """Plugin class, name and provider it comes from."""
269 name: str
270 plugin_class: str
271 provider_name: str
274class HookInfo(NamedTuple):
275 """Hook information."""
277 hook_class_name: str
278 connection_id_attribute_name: str
279 package_name: str
280 hook_name: str
281 connection_type: str
282 connection_testable: bool
283 dialects: list[str] = []
286class ConnectionTypeHookUIMetadata(NamedTuple):
287 """Hook metadata for one connection type (connection UI); ``field_behaviour`` is standard fields."""
289 connection_type: str
290 hook_name: str
291 hook_class_name: str | None
292 field_behaviour: dict | None
295class ConnectionFormWidgetInfo(NamedTuple):
296 """Connection Form Widget information."""
298 hook_class_name: str
299 package_name: str
300 field: Any
301 field_name: str
302 is_sensitive: bool
305def log_optional_feature_disabled(class_name, e, provider_package):
306 """Log optional feature disabled."""
307 log.debug(
308 "Optional feature disabled on exception when importing '%s' from '%s' package",
309 class_name,
310 provider_package,
311 exc_info=e,
312 )
313 log.info(
314 "Optional provider feature disabled when importing '%s' from '%s' package",
315 class_name,
316 provider_package,
317 )
320def log_import_warning(class_name, e, provider_package):
321 """Log import warning."""
322 log.warning(
323 "Exception when importing '%s' from '%s' package",
324 class_name,
325 provider_package,
326 exc_info=e,
327 )
330# This is a temporary measure until all community providers will add AirflowOptionalProviderFeatureException
331# where they have optional features. We are going to add tests in our CI to catch all such cases and will
332# fix them, but until now all "known unhandled optional feature errors" from community providers
333# should be added here
334KNOWN_UNHANDLED_OPTIONAL_FEATURE_ERRORS = [("apache-airflow-providers-google", "No module named 'paramiko'")]
337def _correctness_check(provider_package: str, class_name: str, provider_info: ProviderInfo) -> Any:
338 """
339 Perform coherence check on provider classes.
341 For apache-airflow providers - it checks if it starts with appropriate package. For all providers
342 it tries to import the provider - checking that there are no exceptions during importing.
343 It logs appropriate warning in case it detects any problems.
345 :param provider_package: name of the provider package
346 :param class_name: name of the class to import
348 :return the class if the class is OK, None otherwise.
349 """
350 if not _check_builtin_provider_prefix(provider_package, class_name):
351 return None
352 try:
353 imported_class = import_string(class_name)
354 except AirflowOptionalProviderFeatureException as e:
355 # When the provider class raises AirflowOptionalProviderFeatureException
356 # this is an expected case when only some classes in provider are
357 # available. We just log debug level here and print info message in logs so that
358 # the user is aware of it
359 log_optional_feature_disabled(class_name, e, provider_package)
360 return None
361 except ImportError as e:
362 if "No module named 'airflow.providers." in e.msg:
363 # handle cases where another provider is missing. This can only happen if
364 # there is an optional feature, so we log debug and print information about it
365 log_optional_feature_disabled(class_name, e, provider_package)
366 return None
367 for known_error in KNOWN_UNHANDLED_OPTIONAL_FEATURE_ERRORS:
368 # Until we convert all providers to use AirflowOptionalProviderFeatureException
369 # we assume any problem with importing another "provider" is because this is an
370 # optional feature, so we log debug and print information about it
371 if known_error[0] == provider_package and known_error[1] in e.msg:
372 log_optional_feature_disabled(class_name, e, provider_package)
373 return None
374 # But when we have no idea - we print warning to logs
375 log_import_warning(class_name, e, provider_package)
376 return None
377 except Exception as e:
378 log_import_warning(class_name, e, provider_package)
379 return None
380 return imported_class
383# We want to have better control over initialization of parameters and be able to debug and test it
384# So we add our own decorator
385def provider_info_cache(cache_name: str) -> Callable[[Callable[PS, None]], Callable[PS, None]]:
386 """
387 Decorate and cache provider info.
389 Decorator factory that create decorator that caches initialization of provider's parameters
390 :param cache_name: Name of the cache
391 """
393 def provider_info_cache_decorator(func: Callable[PS, None]) -> Callable[PS, None]:
394 @wraps(func)
395 def wrapped_function(*args: PS.args, **kwargs: PS.kwargs) -> None:
396 providers_manager_instance = args[0]
397 if TYPE_CHECKING:
398 assert isinstance(providers_manager_instance, ProvidersManager)
400 if cache_name in providers_manager_instance._initialized_cache:
401 return
402 start_time = perf_counter()
403 log.debug("Initializing Providers Manager[%s]", cache_name)
404 func(*args, **kwargs)
405 providers_manager_instance._initialized_cache[cache_name] = True
406 log.debug(
407 "Initialization of Providers Manager[%s] took %.2f seconds",
408 cache_name,
409 perf_counter() - start_time,
410 )
412 return wrapped_function
414 return provider_info_cache_decorator
417class ProvidersManager(LoggingMixin):
418 """
419 Manages all provider distributions.
421 This is a Singleton class. The first time it is
422 instantiated, it discovers all available providers in installed packages.
423 """
425 resource_version = "0"
426 _initialized: bool = False
427 _initialization_stack_trace = None
428 _instance: ProvidersManager | None = None
430 def __new__(cls):
431 if cls._instance is None:
432 cls._instance = super().__new__(cls)
433 return cls._instance
435 @staticmethod
436 def initialized() -> bool:
437 return ProvidersManager._initialized
439 @staticmethod
440 def initialization_stack_trace() -> str | None:
441 return ProvidersManager._initialization_stack_trace
443 def __init__(self):
444 """Initialize the manager."""
445 # skip initialization if already initialized
446 if self.initialized():
447 return
449 super().__init__()
450 ProvidersManager._initialized = True
451 ProvidersManager._initialization_stack_trace = "".join(traceback.format_stack(inspect.currentframe()))
452 self._initialized_cache: dict[str, bool] = {}
453 # Keeps dict of providers keyed by module name
454 self._provider_dict: dict[str, ProviderInfo] = {}
455 self._fs_set: set[str] = set()
456 self._asset_uri_handlers: dict[str, Callable[[SplitResult], SplitResult]] = {}
457 self._asset_factories: dict[str, Callable[..., Asset]] = {}
458 self._asset_to_openlineage_converters: dict[str, Callable] = {}
459 self._taskflow_decorators: dict[str, Callable] = LazyDictWithCache()
460 # keeps mapping between connection_types and hook class, package they come from
461 self._hook_provider_dict: dict[str, HookClassProvider] = {}
462 self._dialect_provider_dict: dict[str, DialectInfo] = {}
463 # Keeps dict of hooks keyed by connection type. They are lazy evaluated at access time
464 self._hooks_lazy_dict: LazyDictWithCache[str, HookInfo | Callable] = LazyDictWithCache()
465 # Keeps hook display names read from provider.yaml (hook-name field)
466 self._hook_name_dict: dict[str, str] = {}
467 # Keeps methods that should be used to add custom widgets tuple of keyed by name of the extra field
468 self._connection_form_widgets: dict[str, ConnectionFormWidgetInfo] = {}
469 # Customizations for javascript fields are kept here
470 self._field_behaviours: dict[str, dict] = {}
471 self._cli_command_functions_set: set[Callable[[], list[CLICommand]]] = set()
472 self._cli_command_provider_name_set: set[str] = set()
473 self._extra_link_class_name_set: set[str] = set()
474 self._logging_class_name_set: set[str] = set()
475 self._remote_logging_info_list: list[RemoteLoggingInfo] = []
476 self._remote_logging_by_scheme: dict[str, RemoteLoggingInfo] = {}
477 self._auth_manager_class_name_set: set[str] = set()
478 self._auth_manager_without_check_set: set[tuple[str, str]] = set()
479 self._secrets_backend_class_name_set: set[str] = set()
480 self._executor_class_name_set: set[str] = set()
481 self._executor_without_check_set: set[tuple[str, str]] = set()
482 self._queue_class_name_set: set[str] = set()
483 self._db_manager_class_name_set: set[str] = set()
484 self._provider_configs: dict[str, dict[str, Any]] = {}
485 self._trigger_info_set: set[TriggerInfo] = set()
486 self._notification_info_set: set[NotificationInfo] = set()
487 self._provider_schema_validator = _create_provider_info_schema_validator()
488 self._customized_form_fields_schema_validator = (
489 _create_customized_form_field_behaviours_schema_validator()
490 )
491 # Set of plugins contained in providers
492 self._plugins_set: set[PluginInfo] = set()
493 self._init_airflow_core_hooks()
495 self._runtime_manager = None
497 def __getattribute__(self, name: str):
498 # Hacky but does the trick for now
499 runtime_properties = {
500 "hooks",
501 "taskflow_decorators",
502 "filesystem_module_names",
503 "asset_factories",
504 "asset_uri_handlers",
505 "asset_to_openlineage_converters",
506 }
508 if name in runtime_properties:
509 warnings.warn(
510 f"ProvidersManager.{name} is deprecated. Use ProvidersManagerTaskRuntime.{name} from task-sdk instead.",
511 DeprecatedImportWarning,
512 stacklevel=2,
513 )
514 if object.__getattribute__(self, "_runtime_manager") is None:
515 from airflow.sdk.providers_manager_runtime import ProvidersManagerTaskRuntime
517 object.__setattr__(self, "_runtime_manager", ProvidersManagerTaskRuntime())
518 return getattr(object.__getattribute__(self, "_runtime_manager"), name)
520 return object.__getattribute__(self, name)
522 def _init_airflow_core_hooks(self):
523 """Initialize the hooks dict with default hooks from Airflow core."""
524 core_dummy_hooks = {
525 "generic": "Generic",
526 "email": "Email",
527 }
528 for key, display in core_dummy_hooks.items():
529 self._hooks_lazy_dict[key] = HookInfo(
530 hook_class_name=None,
531 connection_id_attribute_name=None,
532 package_name=None,
533 hook_name=display,
534 connection_type=None,
535 connection_testable=False,
536 )
537 for conn_type, class_name in (
538 ("fs", "airflow.providers.standard.hooks.filesystem.FSHook"),
539 ("package_index", "airflow.providers.standard.hooks.package_index.PackageIndexHook"),
540 ):
541 self._hooks_lazy_dict[conn_type] = functools.partial(
542 self._import_hook,
543 connection_type=None,
544 package_name="apache-airflow-providers-standard",
545 hook_class_name=class_name,
546 provider_info=None,
547 )
549 @provider_info_cache("list")
550 def initialize_providers_list(self):
551 """Lazy initialization of providers list."""
552 # Local source folders are loaded first. They should take precedence over the package ones for
553 # Development purpose. In production provider.yaml files are not present in the 'airflow" directory
554 # So there is no risk we are going to override package provider accidentally. This can only happen
555 # in case of local development
556 discover_all_providers_from_packages(
557 self._provider_dict, self._provider_schema_validator, provider_incompatibility_reason
558 )
559 self._verify_all_providers_all_compatible()
560 self._provider_dict = dict(sorted(self._provider_dict.items()))
562 def _verify_all_providers_all_compatible(self):
563 from packaging import version as packaging_version
565 for provider_id, info in self._provider_dict.items():
566 min_version = MIN_PROVIDER_VERSIONS.get(provider_id)
567 if min_version:
568 if packaging_version.parse(min_version) > packaging_version.parse(info.version):
569 log.warning(
570 "The package %s is not compatible with this version of Airflow. "
571 "The package has version %s but the minimum supported version "
572 "of the package is %s",
573 provider_id,
574 info.version,
575 min_version,
576 )
578 @provider_info_cache("hooks")
579 def initialize_providers_hooks(self):
580 """Lazy initialization of providers hooks."""
581 self._init_airflow_core_hooks()
582 self.initialize_providers_list()
583 self._discover_hooks()
584 self._load_ui_metadata()
585 self._hook_provider_dict = dict(sorted(self._hook_provider_dict.items()))
587 @provider_info_cache("filesystems")
588 def initialize_providers_filesystems(self):
589 """Lazy initialization of providers filesystems."""
590 self.initialize_providers_list()
591 self._discover_filesystems()
593 @provider_info_cache("asset_uris")
594 def initialize_providers_asset_uri_resources(self):
595 """Lazy initialization of provider asset URI handlers, factories, converters etc."""
596 self.initialize_providers_list()
597 self._discover_asset_uri_resources()
599 @provider_info_cache("hook_lineage_writers")
600 @provider_info_cache("taskflow_decorators")
601 def initialize_providers_taskflow_decorator(self):
602 """Lazy initialization of providers hooks."""
603 self.initialize_providers_list()
604 self._discover_taskflow_decorators()
606 @provider_info_cache("extra_links")
607 def initialize_providers_extra_links(self):
608 """Lazy initialization of providers extra links."""
609 self.initialize_providers_list()
610 self._discover_extra_links()
612 @provider_info_cache("logging")
613 def initialize_providers_logging(self):
614 """Lazy initialization of providers logging information."""
615 self.initialize_providers_list()
616 self._discover_logging()
618 @provider_info_cache("remote_logging")
619 def initialize_providers_remote_logging(self):
620 """Lazy initialization of providers remote logging IO handlers."""
621 self.initialize_providers_list()
622 self._discover_remote_logging()
624 @provider_info_cache("secrets_backends")
625 def initialize_providers_secrets_backends(self):
626 """Lazy initialization of providers secrets_backends information."""
627 self.initialize_providers_list()
628 self._discover_secrets_backends()
630 @provider_info_cache("executors")
631 def initialize_providers_executors(self):
632 """Lazy initialization of providers executors information."""
633 self.initialize_providers_list()
634 self._discover_executors(check=True)
636 @provider_info_cache("executors_without_check")
637 def initialize_providers_executors_without_check(self):
638 """Lazy initialization of providers executors information."""
639 self.initialize_providers_list()
640 self._discover_executors(check=False)
642 @provider_info_cache("queues")
643 def initialize_providers_queues(self):
644 """Lazy initialization of providers queue information."""
645 self.initialize_providers_list()
646 self._discover_queues()
648 @provider_info_cache("db_managers")
649 def initialize_providers_db_managers(self):
650 """Lazy initialization of providers db_managers information."""
651 self.initialize_providers_list()
652 self._discover_db_managers()
654 @provider_info_cache("notifications")
655 def initialize_providers_notifications(self):
656 """Lazy initialization of providers notifications information."""
657 self.initialize_providers_list()
658 self._discover_notifications()
660 @provider_info_cache("auth_managers")
661 def initialize_providers_auth_managers(self):
662 """Lazy initialization of providers auth manager information."""
663 self.initialize_providers_list()
664 self._discover_auth_managers(check=True)
666 @provider_info_cache("auth_managers_without_check")
667 def initialize_providers_auth_managers_without_check(self):
668 """Lazy initialization of providers auth manager information."""
669 self.initialize_providers_list()
670 self._discover_auth_managers(check=False)
672 @provider_info_cache("config")
673 def initialize_providers_configuration(self):
674 """Lazy initialization of provider configuration metadata and merge it into ``conf``."""
675 self.initialize_providers_list()
676 self._discover_config()
678 @provider_info_cache("plugins")
679 def initialize_providers_plugins(self):
680 self.initialize_providers_list()
681 self._discover_plugins()
683 @provider_info_cache("cli_command")
684 def initialize_providers_cli_command(self):
685 """Lazy initialization of providers CLI commands."""
686 self.initialize_providers_list()
687 self._discover_cli_command()
689 def _discover_hooks_from_connection_types(
690 self,
691 hook_class_names_registered: set[str],
692 already_registered_warning_connection_types: set[str],
693 package_name: str,
694 provider: ProviderInfo,
695 ):
696 """
697 Discover hooks from the "connection-types" property.
699 This is new, better method that replaces discovery from hook-class-names as it
700 allows to lazy import individual Hook classes when they are accessed.
701 The "connection-types" keeps information about both - connection type and class
702 name so we can discover all connection-types without importing the classes.
703 :param hook_class_names_registered: set of registered hook class names for this provider
704 :param already_registered_warning_connection_types: set of connections for which warning should be
705 printed in logs as they were already registered before
706 :param package_name:
707 :param provider:
708 :return:
709 """
710 provider_uses_connection_types = False
711 connection_types = provider.data.get("connection-types")
712 if connection_types:
713 for connection_type_dict in connection_types:
714 connection_type = connection_type_dict["connection-type"]
715 hook_class_name = connection_type_dict["hook-class-name"]
716 hook_class_names_registered.add(hook_class_name)
717 already_registered = self._hook_provider_dict.get(connection_type)
718 if already_registered:
719 if already_registered.package_name != package_name:
720 already_registered_warning_connection_types.add(connection_type)
721 else:
722 log.warning(
723 "The connection type '%s' is already registered in the"
724 " package '%s' with different class names: '%s' and '%s'. ",
725 connection_type,
726 package_name,
727 already_registered.hook_class_name,
728 hook_class_name,
729 )
730 else:
731 self._hook_provider_dict[connection_type] = HookClassProvider(
732 hook_class_name=hook_class_name, package_name=package_name
733 )
734 # Defer importing hook to access time by setting import hook method as dict value
735 self._hooks_lazy_dict[connection_type] = functools.partial(
736 self._import_hook,
737 connection_type=connection_type,
738 provider_info=provider,
739 )
740 provider_uses_connection_types = True
741 return provider_uses_connection_types
743 def _discover_hooks_from_hook_class_names(
744 self,
745 hook_class_names_registered: set[str],
746 already_registered_warning_connection_types: set[str],
747 package_name: str,
748 provider: ProviderInfo,
749 provider_uses_connection_types: bool,
750 ):
751 """
752 Discover hooks from "hook-class-names' property.
754 This property is deprecated but we should support it in Airflow 2.
755 The hook-class-names array contained just Hook names without connection type,
756 therefore we need to import all those classes immediately to know which connection types
757 are supported. This makes it impossible to selectively only import those hooks that are used.
758 :param already_registered_warning_connection_types: list of connection hooks that we should warn
759 about when finished discovery
760 :param package_name: name of the provider package
761 :param provider: class that keeps information about version and details of the provider
762 :param provider_uses_connection_types: determines whether the provider uses "connection-types" new
763 form of passing connection types
764 :return:
765 """
766 hook_class_names = provider.data.get("hook-class-names")
767 if hook_class_names:
768 for hook_class_name in hook_class_names:
769 if hook_class_name in hook_class_names_registered:
770 # Silently ignore the hook class - it's already marked for lazy-import by
771 # connection-types discovery
772 continue
773 hook_info = self._import_hook(
774 connection_type=None,
775 provider_info=provider,
776 hook_class_name=hook_class_name,
777 package_name=package_name,
778 )
779 if not hook_info:
780 # Problem why importing class - we ignore it. Log is written at import time
781 continue
782 already_registered = self._hook_provider_dict.get(hook_info.connection_type)
783 if already_registered:
784 if already_registered.package_name != package_name:
785 already_registered_warning_connection_types.add(hook_info.connection_type)
786 else:
787 if already_registered.hook_class_name != hook_class_name:
788 log.warning(
789 "The hook connection type '%s' is registered twice in the"
790 " package '%s' with different class names: '%s' and '%s'. "
791 " Please fix it!",
792 hook_info.connection_type,
793 package_name,
794 already_registered.hook_class_name,
795 hook_class_name,
796 )
797 else:
798 self._hook_provider_dict[hook_info.connection_type] = HookClassProvider(
799 hook_class_name=hook_class_name, package_name=package_name
800 )
801 self._hooks_lazy_dict[hook_info.connection_type] = hook_info
803 if not provider_uses_connection_types:
804 warnings.warn(
805 f"The provider {package_name} uses `hook-class-names` "
806 "property in provider-info and has no `connection-types` one. "
807 "The 'hook-class-names' property has been deprecated in favour "
808 "of 'connection-types' in Airflow 2.2. Use **both** in case you want to "
809 "have backwards compatibility with Airflow < 2.2",
810 DeprecationWarning,
811 stacklevel=1,
812 )
813 for already_registered_connection_type in already_registered_warning_connection_types:
814 log.warning(
815 "The connection_type '%s' has been already registered by provider '%s.'",
816 already_registered_connection_type,
817 self._hook_provider_dict[already_registered_connection_type].package_name,
818 )
820 def _discover_hooks(self) -> None:
821 """Retrieve all connections defined in the providers via Hooks."""
822 for package_name, provider in self._provider_dict.items():
823 duplicated_connection_types: set[str] = set()
824 hook_class_names_registered: set[str] = set()
825 self._discover_provider_dialects(package_name, provider)
826 provider_uses_connection_types = self._discover_hooks_from_connection_types(
827 hook_class_names_registered, duplicated_connection_types, package_name, provider
828 )
829 self._discover_hooks_from_hook_class_names(
830 hook_class_names_registered,
831 duplicated_connection_types,
832 package_name,
833 provider,
834 provider_uses_connection_types,
835 )
836 self._hook_provider_dict = dict(sorted(self._hook_provider_dict.items()))
838 def _discover_provider_dialects(self, provider_name: str, provider: ProviderInfo):
839 dialects = provider.data.get("dialects", [])
840 if dialects:
841 self._dialect_provider_dict.update(
842 {
843 item["dialect-type"]: DialectInfo(
844 name=item["dialect-type"],
845 dialect_class_name=item["dialect-class-name"],
846 provider_name=provider_name,
847 )
848 for item in dialects
849 }
850 )
852 @provider_info_cache("import_all_hooks")
853 def _import_info_from_all_hooks(self):
854 """Force-import all hooks and initialize the connections/fields."""
855 # Retrieve all hooks to make sure that all of them are imported
856 _ = list(self._hooks_lazy_dict.values())
857 self._field_behaviours = dict(sorted(self._field_behaviours.items()))
859 # Widgets for connection forms are currently used in two places:
860 # 1. In the UI Connections, expected same order that it defined in Hook.
861 # 2. cli command - `airflow providers widgets` and expected that it in alphabetical order.
862 # It is not possible to recover original ordering after sorting,
863 # that the main reason why original sorting moved to cli part:
864 # self._connection_form_widgets = dict(sorted(self._connection_form_widgets.items()))
866 def _discover_filesystems(self) -> None:
867 """Retrieve all filesystems defined in the providers."""
868 for provider_package, provider in self._provider_dict.items():
869 for fs_module_name in provider.data.get("filesystems", []):
870 if _correctness_check(provider_package, f"{fs_module_name}.get_fs", provider):
871 self._fs_set.add(fs_module_name)
872 self._fs_set = set(sorted(self._fs_set))
874 def _discover_asset_uri_resources(self) -> None:
875 """Discovers and registers asset URI handlers, factories, and converters for all providers."""
876 from airflow.sdk.definitions.asset import normalize_noop
878 def _safe_register_resource(
879 provider_package_name: str,
880 schemes_list: list[str],
881 resource_path: str | None,
882 resource_registry: dict,
883 default_resource: Any = None,
884 ):
885 """
886 Register a specific resource (handler, factory, or converter) for the given schemes.
888 If the resolved resource (either from the path or the default) is valid, it updates
889 the resource registry with the appropriate resource for each scheme.
890 """
891 resource = (
892 _correctness_check(provider_package_name, resource_path, provider)
893 if resource_path is not None
894 else default_resource
895 )
896 if resource:
897 resource_registry.update((scheme, resource) for scheme in schemes_list)
899 for provider_name, provider in self._provider_dict.items():
900 for uri_info in provider.data.get("asset-uris", []):
901 if "schemes" not in uri_info or "handler" not in uri_info:
902 continue # Both schemas and handler must be explicitly set, handler can be set to null
903 common_args = {"schemes_list": uri_info["schemes"], "provider_package_name": provider_name}
904 _safe_register_resource(
905 resource_path=uri_info["handler"],
906 resource_registry=self._asset_uri_handlers,
907 default_resource=normalize_noop,
908 **common_args,
909 )
910 _safe_register_resource(
911 resource_path=uri_info.get("factory"),
912 resource_registry=self._asset_factories,
913 **common_args,
914 )
915 _safe_register_resource(
916 resource_path=uri_info.get("to_openlineage_converter"),
917 resource_registry=self._asset_to_openlineage_converters,
918 **common_args,
919 )
921 def _discover_taskflow_decorators(self) -> None:
922 for name, info in self._provider_dict.items():
923 for taskflow_decorator in info.data.get("task-decorators", []):
924 self._add_taskflow_decorator(
925 taskflow_decorator["name"], taskflow_decorator["class-name"], name
926 )
928 def _add_taskflow_decorator(self, name, decorator_class_name: str, provider_package: str) -> None:
929 if not _check_builtin_provider_prefix(provider_package, decorator_class_name):
930 return
932 if name in self._taskflow_decorators:
933 try:
934 existing = self._taskflow_decorators[name]
935 other_name = f"{existing.__module__}.{existing.__name__}"
936 except Exception:
937 # If problem importing, then get the value from the functools.partial
938 other_name = self._taskflow_decorators._raw_dict[name].args[0] # type: ignore[attr-defined]
940 log.warning(
941 "The taskflow decorator '%s' has been already registered (by %s).",
942 name,
943 other_name,
944 )
945 return
947 self._taskflow_decorators[name] = functools.partial(import_string, decorator_class_name)
949 @staticmethod
950 def _get_attr(obj: Any, attr_name: str):
951 """Retrieve attributes of an object, or warn if not found."""
952 if not hasattr(obj, attr_name):
953 log.warning("The object '%s' is missing %s attribute and cannot be registered", obj, attr_name)
954 return None
955 return getattr(obj, attr_name)
957 def _get_connection_type_config(self, provider_info: ProviderInfo, connection_type: str) -> dict | None:
958 """Get connection type config from provider.yaml if it exists."""
959 connection_types = provider_info.data.get("connection-types", [])
960 for conn_config in connection_types:
961 if conn_config.get("connection-type") == connection_type:
962 return conn_config
963 return None
965 def _to_api_format(self, field_name: str, field_def: dict) -> dict:
966 """Convert conn-fields definition to format expected by the API."""
967 schema_def = field_def.get("schema", {})
969 # build schema dict with label moved to `title` per jsonschema convention
970 schema = schema_def.copy()
971 if "label" in field_def:
972 schema["title"] = field_def.get("label")
974 return {
975 "value": schema_def.get("default"),
976 "schema": schema,
977 "description": field_def.get("description"),
978 "source": None,
979 }
981 def _add_widgets(
982 self, package_name: str, hook_class_name: str, connection_type: str, conn_fields: dict
983 ) -> None:
984 """Parse conn-fields from provider info and add to connection_form_widgets."""
985 for field_name, field_def in conn_fields.items():
986 field_data = self._to_api_format(field_name, field_def)
988 prefixed_name = f"extra__{connection_type}__{field_name}"
989 if prefixed_name in self._connection_form_widgets:
990 log.warning(
991 "Field %s for connection type %s already added, skipping",
992 field_name,
993 connection_type,
994 )
995 continue
997 schema_def = field_def.get("schema", {})
998 self._connection_form_widgets[prefixed_name] = ConnectionFormWidgetInfo(
999 hook_class_name=hook_class_name,
1000 package_name=package_name,
1001 field=field_data,
1002 field_name=field_name,
1003 is_sensitive=schema_def.get("format") == "password",
1004 )
1006 def _add_customized_fields(self, package_name: str, connection_type: str, behaviour: dict) -> None:
1007 """Process ui-field-behaviour from provider info and add to field_behaviours."""
1008 if connection_type in self._field_behaviours:
1009 log.warning(
1010 "Field behaviour for connection type %s already exists, skipping",
1011 connection_type,
1012 )
1013 return
1015 # convert kebab-case keys to python style
1016 customized_fields = {
1017 "hidden_fields": behaviour.get("hidden-fields", []),
1018 "relabeling": behaviour.get("relabeling", {}),
1019 "placeholders": behaviour.get("placeholders", {}),
1020 }
1022 try:
1023 self._customized_form_fields_schema_validator.validate(customized_fields)
1024 customized_fields = _ensure_prefix_for_placeholders(customized_fields, connection_type)
1025 self._field_behaviours[connection_type] = customized_fields
1026 except Exception as e:
1027 log.warning(
1028 "Failed to add field behaviour for %s in package %s: %s",
1029 connection_type,
1030 package_name,
1031 e,
1032 )
1034 def _load_ui_metadata(self) -> None:
1035 """Load connection form UI metadata from provider info without importing hooks."""
1036 for package_name, provider in self._provider_dict.items():
1037 for conn_config in provider.data.get("connection-types", []):
1038 connection_type = conn_config.get("connection-type")
1039 hook_class_name = conn_config.get("hook-class-name")
1040 if not connection_type or not hook_class_name:
1041 continue
1043 if hook_name := conn_config.get("hook-name"):
1044 self._hook_name_dict[connection_type] = hook_name
1046 if conn_fields := conn_config.get("conn-fields"):
1047 self._add_widgets(package_name, hook_class_name, connection_type, conn_fields)
1049 if behaviour := conn_config.get("ui-field-behaviour"):
1050 self._add_customized_fields(package_name, connection_type, behaviour)
1052 def _import_hook(
1053 self,
1054 connection_type: str | None,
1055 provider_info: ProviderInfo,
1056 hook_class_name: str | None = None,
1057 package_name: str | None = None,
1058 ) -> HookInfo | None:
1059 """
1060 Import hook and retrieve hook information.
1062 Either connection_type (for lazy loading) or hook_class_name must be set - but not both).
1063 Only needs package_name if hook_class_name is passed (for lazy loading, package_name
1064 is retrieved from _connection_type_class_provider_dict together with hook_class_name).
1066 :param connection_type: type of the connection
1067 :param hook_class_name: name of the hook class
1068 :param package_name: provider package - only needed in case connection_type is missing
1069 : return
1070 """
1071 if connection_type is None and hook_class_name is None:
1072 raise ValueError("Either connection_type or hook_class_name must be set")
1073 if connection_type is not None and hook_class_name is not None:
1074 raise ValueError(
1075 f"Both connection_type ({connection_type} and "
1076 f"hook_class_name {hook_class_name} are set. Only one should be set!"
1077 )
1078 if connection_type is not None:
1079 class_provider = self._hook_provider_dict[connection_type]
1080 package_name = class_provider.package_name
1081 hook_class_name = class_provider.hook_class_name
1082 else:
1083 if not hook_class_name:
1084 raise ValueError("Either connection_type or hook_class_name must be set")
1085 if not package_name:
1086 raise ValueError(
1087 f"Provider package name is not set when hook_class_name ({hook_class_name}) is used"
1088 )
1089 hook_class: type[BaseHook] | None = _correctness_check(package_name, hook_class_name, provider_info)
1090 if hook_class is None:
1091 return None
1093 # Check if provider info already has UI metadata and skip Python hook methods
1094 # to avoid duplicate initialization and unnecessary wtforms imports
1095 ui_metadata_loaded = False
1096 if provider_info and connection_type:
1097 conn_config = self._get_connection_type_config(provider_info, connection_type)
1098 ui_metadata_loaded = conn_config is not None and bool(
1099 conn_config.get("conn-fields") or conn_config.get("ui-field-behaviour")
1100 )
1102 if not ui_metadata_loaded:
1103 try:
1104 from wtforms import BooleanField, IntegerField, PasswordField, StringField
1106 allowed_field_classes = [IntegerField, PasswordField, StringField, BooleanField]
1107 # Do not use attr here. We want to check only direct class fields not those
1108 # inherited from parent hook. This way we add form fields only once for the whole
1109 # hierarchy and we add it only from the parent hook that provides those!
1110 if "get_connection_form_widgets" in hook_class.__dict__:
1111 warning = AirflowProviderDeprecationWarning(
1112 f"Hook '{hook_class_name}' defines get_connection_form_widgets(). "
1113 "This method is deprecated. Define connection fields declaratively in "
1114 "provider.yaml under 'conn-fields' instead. See "
1115 "https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html"
1116 )
1117 warning.deprecated_provider_since = "3.2.0"
1118 warnings.warn(warning, stacklevel=2)
1119 widgets = hook_class.get_connection_form_widgets()
1120 if widgets:
1121 for widget in widgets.values():
1122 if widget.field_class not in allowed_field_classes:
1123 log.warning(
1124 "The hook_class '%s' uses field of unsupported class '%s'. "
1125 "Only '%s' field classes are supported",
1126 hook_class_name,
1127 widget.field_class,
1128 allowed_field_classes,
1129 )
1130 return None
1131 self._add_widgets_from_hook(package_name, hook_class, widgets)
1132 if "get_ui_field_behaviour" in hook_class.__dict__:
1133 warning = AirflowProviderDeprecationWarning(
1134 f"Hook '{hook_class_name}' defines get_ui_field_behaviour(). "
1135 "This method is deprecated. Define field behaviour declaratively in "
1136 "provider.yaml under 'ui-field-behaviour' instead. See "
1137 "https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html"
1138 )
1139 warning.deprecated_provider_since = "3.2.0"
1140 warnings.warn(warning, stacklevel=2)
1141 field_behaviours = hook_class.get_ui_field_behaviour()
1142 if field_behaviours:
1143 self._add_customized_fields_from_hook(package_name, hook_class, field_behaviours)
1144 except ImportError as e:
1145 if e.name in ["flask_appbuilder", "wtforms"]:
1146 log.info(
1147 "The hook_class '%s' is not fully initialized (UI widgets will be missing), because "
1148 "the 'flask_appbuilder' package is not installed, however it is not required for "
1149 "Airflow components to work",
1150 hook_class_name,
1151 )
1152 except Exception as e:
1153 log.warning(
1154 "Exception when importing '%s' from '%s' package: %s",
1155 hook_class_name,
1156 package_name,
1157 e,
1158 )
1159 return None
1161 hook_connection_type = self._get_attr(hook_class, "conn_type")
1162 if connection_type:
1163 if hook_connection_type != connection_type:
1164 log.warning(
1165 "Inconsistency! The hook class '%s' declares connection type '%s'"
1166 " but it is added by provider '%s' as connection_type '%s' in provider info. "
1167 "This should be fixed!",
1168 hook_class,
1169 hook_connection_type,
1170 package_name,
1171 connection_type,
1172 )
1173 connection_type = hook_connection_type
1174 connection_id_attribute_name: str = self._get_attr(hook_class, "conn_name_attr")
1175 hook_name: str = self._get_attr(hook_class, "hook_name")
1177 if not connection_type or not connection_id_attribute_name or not hook_name:
1178 log.warning(
1179 "The hook misses one of the key attributes: "
1180 "conn_type: %s, conn_id_attribute_name: %s, hook_name: %s",
1181 connection_type,
1182 connection_id_attribute_name,
1183 hook_name,
1184 )
1185 return None
1187 return HookInfo(
1188 hook_class_name=hook_class_name,
1189 connection_id_attribute_name=connection_id_attribute_name,
1190 package_name=package_name,
1191 hook_name=hook_name,
1192 connection_type=connection_type,
1193 connection_testable=hasattr(hook_class, "test_connection"),
1194 )
1196 def _add_widgets_from_hook(self, package_name: str, hook_class: type, widgets: dict[str, Any]):
1197 conn_type = hook_class.conn_type # type: ignore
1198 for field_identifier, field in widgets.items():
1199 if field_identifier.startswith("extra__"):
1200 prefixed_field_name = field_identifier
1201 else:
1202 prefixed_field_name = f"extra__{conn_type}__{field_identifier}"
1203 if prefixed_field_name in self._connection_form_widgets:
1204 log.warning(
1205 "The field %s from class %s has already been added by another provider. Ignoring it.",
1206 field_identifier,
1207 hook_class.__name__,
1208 )
1209 # In case of inherited hooks this might be happening several times
1210 else:
1211 self._connection_form_widgets[prefixed_field_name] = ConnectionFormWidgetInfo(
1212 hook_class.__name__,
1213 package_name,
1214 field,
1215 field_identifier,
1216 hasattr(field.field_class.widget, "input_type")
1217 and field.field_class.widget.input_type == "password",
1218 )
1220 def _add_customized_fields_from_hook(self, package_name: str, hook_class: type, customized_fields: dict):
1221 try:
1222 connection_type = getattr(hook_class, "conn_type")
1224 self._customized_form_fields_schema_validator.validate(customized_fields)
1226 if connection_type:
1227 customized_fields = _ensure_prefix_for_placeholders(customized_fields, connection_type)
1229 if connection_type in self._field_behaviours:
1230 log.warning(
1231 "The connection_type %s from package %s and class %s has already been added "
1232 "by another provider. Ignoring it.",
1233 connection_type,
1234 package_name,
1235 hook_class.__name__,
1236 )
1237 return
1238 self._field_behaviours[connection_type] = customized_fields
1239 except Exception as e:
1240 log.warning(
1241 "Error when loading customized fields from package '%s' hook class '%s': %s",
1242 package_name,
1243 hook_class.__name__,
1244 e,
1245 )
1247 def _discover_auth_managers(self, *, check: bool) -> None:
1248 """Retrieve all auth managers defined in the providers."""
1249 for provider_package, provider in self._provider_dict.items():
1250 if provider.data.get("auth-managers"):
1251 for auth_manager_class_name in provider.data["auth-managers"]:
1252 if not check:
1253 self._auth_manager_without_check_set.add((auth_manager_class_name, provider_package))
1254 elif _correctness_check(provider_package, auth_manager_class_name, provider):
1255 self._auth_manager_class_name_set.add(auth_manager_class_name)
1257 def _discover_cli_command(self) -> None:
1258 """Retrieve all CLI command functions defined in the providers."""
1259 for provider_package, provider in self._provider_dict.items():
1260 if provider.data.get("cli"):
1261 for cli_command_function_name in provider.data["cli"]:
1262 # _correctness_check will return the function if found and correct
1263 # we store the function itself instead of its name to avoid importing it again later in cli_parser to speed up cli loading
1264 if cli_func := _correctness_check(provider_package, cli_command_function_name, provider):
1265 cli_func = cast("Callable[[], list[CLICommand]]", cli_func)
1266 self._cli_command_functions_set.add(cli_func)
1267 self._cli_command_provider_name_set.add(provider_package)
1269 def _discover_notifications(self) -> None:
1270 """Retrieve all notifications defined in the providers."""
1271 for provider_package, provider in self._provider_dict.items():
1272 if provider.data.get("notifications"):
1273 for notification_class_name in provider.data["notifications"]:
1274 if _correctness_check(provider_package, notification_class_name, provider):
1275 self._notification_info_set.add(notification_class_name)
1277 def _discover_extra_links(self) -> None:
1278 """Retrieve all extra links defined in the providers."""
1279 for provider_package, provider in self._provider_dict.items():
1280 if provider.data.get("extra-links"):
1281 for extra_link_class_name in provider.data["extra-links"]:
1282 if _correctness_check(provider_package, extra_link_class_name, provider):
1283 self._extra_link_class_name_set.add(extra_link_class_name)
1285 def _discover_logging(self) -> None:
1286 """Retrieve all logging defined in the providers."""
1287 for provider_package, provider in self._provider_dict.items():
1288 if provider.data.get("logging"):
1289 for logging_class_name in provider.data["logging"]:
1290 if _correctness_check(provider_package, logging_class_name, provider):
1291 self._logging_class_name_set.add(logging_class_name)
1293 def _discover_remote_logging(self) -> None:
1294 """Retrieve all remote logging IO handlers defined in the providers."""
1295 for provider_package, provider in self._provider_dict.items():
1296 entries = provider.data.get("remote-logging") or []
1297 for entry in entries:
1298 classpath = entry["classpath"]
1299 if not _correctness_check(provider_package, classpath, provider):
1300 continue
1301 info = RemoteLoggingInfo(
1302 classpath=classpath,
1303 scheme=entry["scheme"],
1304 package_name=provider_package,
1305 )
1306 if (existing := self._remote_logging_by_scheme.get(info.scheme)) is not None:
1307 log.warning(
1308 "Remote logging scheme '%s' is already registered by %s; ignoring "
1309 "duplicate registration from %s.",
1310 info.scheme,
1311 existing.package_name,
1312 info.package_name,
1313 )
1314 continue
1315 self._remote_logging_info_list.append(info)
1316 self._remote_logging_by_scheme[info.scheme] = info
1318 def _discover_secrets_backends(self) -> None:
1319 """Retrieve all secrets backends defined in the providers."""
1320 for provider_package, provider in self._provider_dict.items():
1321 if provider.data.get("secrets-backends"):
1322 for secrets_backends_class_name in provider.data["secrets-backends"]:
1323 if _correctness_check(provider_package, secrets_backends_class_name, provider):
1324 self._secrets_backend_class_name_set.add(secrets_backends_class_name)
1326 def _discover_executors(self, *, check: bool) -> None:
1327 """Retrieve all executors defined in the providers."""
1328 for provider_package, provider in self._provider_dict.items():
1329 if provider.data.get("executors"):
1330 for executors_class_path in provider.data["executors"]:
1331 if not check:
1332 self._executor_without_check_set.add((executors_class_path, provider_package))
1333 elif _correctness_check(provider_package, executors_class_path, provider):
1334 self._executor_class_name_set.add(executors_class_path)
1336 def _discover_queues(self) -> None:
1337 """Retrieve all queues defined in the providers."""
1338 for provider_package, provider in self._provider_dict.items():
1339 if provider.data.get("queues"):
1340 for queue_class_name in provider.data["queues"]:
1341 if _correctness_check(provider_package, queue_class_name, provider):
1342 self._queue_class_name_set.add(queue_class_name)
1344 def _discover_db_managers(self) -> None:
1345 """Retrieve all DB managers defined in the providers."""
1346 for provider_package, provider in self._provider_dict.items():
1347 if provider.data.get("db-managers"):
1348 for db_manager_class_name in provider.data["db-managers"]:
1349 if _correctness_check(provider_package, db_manager_class_name, provider):
1350 self._db_manager_class_name_set.add(db_manager_class_name)
1352 def _discover_config(self) -> None:
1353 """Retrieve all configs defined in the providers."""
1354 for provider_package, provider in self._provider_dict.items():
1355 if provider.data.get("config"):
1356 self._provider_configs[provider_package] = provider.data.get("config") # type: ignore[assignment]
1358 def _discover_plugins(self) -> None:
1359 """Retrieve all plugins defined in the providers."""
1360 for provider_package, provider in self._provider_dict.items():
1361 for plugin_dict in provider.data.get("plugins", ()):
1362 if not _correctness_check(provider_package, plugin_dict["plugin-class"], provider):
1363 log.warning("Plugin not loaded due to above correctness check problem.")
1364 continue
1365 self._plugins_set.add(
1366 PluginInfo(
1367 name=plugin_dict["name"],
1368 plugin_class=plugin_dict["plugin-class"],
1369 provider_name=provider_package,
1370 )
1371 )
1373 @provider_info_cache("triggers")
1374 def initialize_providers_triggers(self):
1375 """Initialize providers triggers."""
1376 self.initialize_providers_list()
1377 for provider_package, provider in self._provider_dict.items():
1378 for trigger in provider.data.get("triggers", []):
1379 for trigger_class_name in trigger.get("python-modules"):
1380 self._trigger_info_set.add(
1381 TriggerInfo(
1382 package_name=provider_package,
1383 trigger_class_name=trigger_class_name,
1384 integration_name=trigger.get("integration-name", ""),
1385 )
1386 )
1388 @property
1389 def auth_managers(self) -> list[str]:
1390 """Returns information about available providers notifications class."""
1391 self.initialize_providers_auth_managers()
1392 return sorted(self._auth_manager_class_name_set)
1394 @property
1395 def auth_manager_without_check(self) -> set[tuple[str, str]]:
1396 """Returns set of (auth manager class names, provider package name) without correctness check."""
1397 self.initialize_providers_auth_managers_without_check()
1398 return self._auth_manager_without_check_set
1400 @property
1401 def cli_command_functions(self) -> set[Callable[[], list[CLICommand]]]:
1402 """Returns list of CLI command function names from providers."""
1403 self.initialize_providers_cli_command()
1404 return self._cli_command_functions_set
1406 @property
1407 def cli_command_providers(self) -> set[str]:
1408 """Returns set of provider package names that provide CLI commands."""
1409 self.initialize_providers_cli_command()
1410 return self._cli_command_provider_name_set
1412 @property
1413 def notification(self) -> list[NotificationInfo]:
1414 """Returns information about available providers notifications class."""
1415 self.initialize_providers_notifications()
1416 return sorted(self._notification_info_set)
1418 @property
1419 def trigger(self) -> list[TriggerInfo]:
1420 """Returns information about available providers trigger class."""
1421 self.initialize_providers_triggers()
1422 return sorted(self._trigger_info_set, key=lambda x: x.package_name)
1424 @property
1425 def providers(self) -> dict[str, ProviderInfo]:
1426 """Returns information about available providers."""
1427 self.initialize_providers_list()
1428 return self._provider_dict
1430 @property
1431 def hooks(self) -> MutableMapping[str, HookInfo | None]:
1432 """
1433 Return dictionary of connection_type-to-hook mapping.
1435 Note that the dict can contain None values if a hook discovered cannot be imported!
1436 """
1437 self.initialize_providers_hooks()
1438 # When we return hooks here it will only be used to retrieve hook information
1439 return self._hooks_lazy_dict
1441 def iter_connection_type_hook_ui_metadata(self) -> Iterator[ConnectionTypeHookUIMetadata]:
1442 """
1443 Yield hook metadata per connection type for the connection UI.
1445 Does not import hook classes.
1446 """
1447 self.initialize_providers_hooks()
1448 all_types = frozenset(self._hooks_lazy_dict) | frozenset(self._hook_provider_dict)
1449 for conn_type in sorted(all_types):
1450 raw_entry = self._hooks_lazy_dict._raw_dict.get(conn_type)
1451 provider_entry = self._hook_provider_dict.get(conn_type)
1452 if isinstance(raw_entry, HookInfo):
1453 hook_name = raw_entry.hook_name
1454 hook_class_name = raw_entry.hook_class_name
1455 elif provider_entry:
1456 hook_name = self._hook_name_dict.get(conn_type, conn_type)
1457 hook_class_name = provider_entry.hook_class_name
1458 else:
1459 hook_name = self._hook_name_dict.get(conn_type, conn_type)
1460 hook_class_name = None
1461 yield ConnectionTypeHookUIMetadata(
1462 connection_type=conn_type,
1463 hook_name=hook_name,
1464 hook_class_name=hook_class_name,
1465 field_behaviour=self._field_behaviours.get(conn_type),
1466 )
1468 @property
1469 def _connection_form_widgets_from_metadata(self) -> dict[str, ConnectionFormWidgetInfo]:
1470 """Return connection form widgets from metadata without importing every hook."""
1471 self.initialize_providers_hooks()
1472 return self._connection_form_widgets
1474 @property
1475 def _field_behaviours_from_metadata(self) -> dict[str, dict]:
1476 """Return field behaviour dicts from metadata without importing every hook."""
1477 self.initialize_providers_hooks()
1478 return self._field_behaviours
1480 @property
1481 def dialects(self) -> MutableMapping[str, DialectInfo]:
1482 """Return dictionary of connection_type-to-dialect mapping."""
1483 self.initialize_providers_hooks()
1484 # When we return dialects here it will only be used to retrieve dialect information
1485 return self._dialect_provider_dict
1487 @property
1488 def plugins(self) -> list[PluginInfo]:
1489 """Returns information about plugins available in providers."""
1490 self.initialize_providers_plugins()
1491 return sorted(self._plugins_set, key=lambda x: x.plugin_class)
1493 @property
1494 def taskflow_decorators(self) -> dict[str, TaskDecorator]:
1495 self.initialize_providers_taskflow_decorator()
1496 return self._taskflow_decorators # type: ignore[return-value]
1498 @property
1499 def extra_links_class_names(self) -> list[str]:
1500 """Returns set of extra link class names."""
1501 self.initialize_providers_extra_links()
1502 return sorted(self._extra_link_class_name_set)
1504 @property
1505 def connection_form_widgets(self) -> dict[str, ConnectionFormWidgetInfo]:
1506 """
1507 Returns widgets for connection forms.
1509 Dictionary keys in the same order that it defined in Hook.
1510 """
1511 self.initialize_providers_hooks()
1512 self._import_info_from_all_hooks()
1513 return self._connection_form_widgets
1515 @property
1516 def field_behaviours(self) -> dict[str, dict]:
1517 """Returns dictionary with field behaviours for connection types."""
1518 self.initialize_providers_hooks()
1519 self._import_info_from_all_hooks()
1520 return self._field_behaviours
1522 @property
1523 def logging_class_names(self) -> list[str]:
1524 """Returns set of log task handlers class names."""
1525 self.initialize_providers_logging()
1526 return sorted(self._logging_class_name_set)
1528 @property
1529 def remote_logging_handlers(self) -> list[RemoteLoggingInfo]:
1530 """Return all remote logging IO handlers contributed by providers."""
1531 self.initialize_providers_remote_logging()
1532 return list(self._remote_logging_info_list)
1534 def remote_logging_handler_by_scheme(self, scheme: str) -> RemoteLoggingInfo | None:
1535 """Return the remote logging IO handler registered for the given URL scheme, if any."""
1536 self.initialize_providers_remote_logging()
1537 return self._remote_logging_by_scheme.get(scheme)
1539 @property
1540 def secrets_backend_class_names(self) -> list[str]:
1541 """Returns set of secret backend class names."""
1542 self.initialize_providers_secrets_backends()
1543 return sorted(self._secrets_backend_class_name_set)
1545 @property
1546 def executor_class_names(self) -> list[str]:
1547 self.initialize_providers_executors()
1548 return sorted(self._executor_class_name_set)
1550 @property
1551 def executor_without_check(self) -> set[tuple[str, str]]:
1552 """Returns set of (executor class names, provider package name) without correctness check."""
1553 self.initialize_providers_executors_without_check()
1554 return self._executor_without_check_set
1556 @property
1557 def queue_class_names(self) -> list[str]:
1558 self.initialize_providers_queues()
1559 return sorted(self._queue_class_name_set)
1561 @property
1562 def db_managers(self) -> list[str]:
1563 self.initialize_providers_db_managers()
1564 return sorted(self._db_manager_class_name_set)
1566 @property
1567 def filesystem_module_names(self) -> list[str]:
1568 self.initialize_providers_filesystems()
1569 return sorted(self._fs_set)
1571 @property
1572 def asset_factories(self) -> dict[str, Callable[..., Asset]]:
1573 self.initialize_providers_asset_uri_resources()
1574 return self._asset_factories
1576 @property
1577 def asset_uri_handlers(self) -> dict[str, Callable[[SplitResult], SplitResult]]:
1578 self.initialize_providers_asset_uri_resources()
1579 return self._asset_uri_handlers
1581 @property
1582 def asset_to_openlineage_converters(
1583 self,
1584 ) -> dict[str, Callable]:
1585 self.initialize_providers_asset_uri_resources()
1586 return self._asset_to_openlineage_converters
1588 @property
1589 def provider_configs(self) -> list[tuple[str, dict[str, Any]]]:
1590 self.initialize_providers_configuration()
1591 return sorted(self._provider_configs.items(), key=lambda x: x[0])
1593 @property
1594 def already_initialized_provider_configs(self) -> list[tuple[str, dict[str, Any]]]:
1595 """
1596 Return provider configs that have already been initialized.
1598 .. deprecated:: 3.2.0
1599 Use ``provider_configs`` instead. This property is kept for backwards
1600 compatibility and will be removed in a future version.
1601 """
1602 warnings.warn(
1603 "already_initialized_provider_configs is deprecated. Use `provider_configs` instead.",
1604 DeprecationWarning,
1605 stacklevel=2,
1606 )
1607 return sorted(self._provider_configs.items(), key=lambda x: x[0])
1609 def _cleanup(self):
1610 self._initialized_cache.clear()
1611 self._provider_dict.clear()
1612 self._fs_set.clear()
1613 self._taskflow_decorators.clear()
1614 self._hook_provider_dict.clear()
1615 self._dialect_provider_dict.clear()
1616 self._hooks_lazy_dict.clear()
1617 self._connection_form_widgets.clear()
1618 self._field_behaviours.clear()
1619 self._extra_link_class_name_set.clear()
1620 self._logging_class_name_set.clear()
1621 self._remote_logging_info_list.clear()
1622 self._remote_logging_by_scheme.clear()
1623 self._auth_manager_class_name_set.clear()
1624 self._auth_manager_without_check_set.clear()
1625 self._secrets_backend_class_name_set.clear()
1626 self._executor_class_name_set.clear()
1627 self._executor_without_check_set.clear()
1628 self._queue_class_name_set.clear()
1629 self._provider_configs.clear()
1631 # Imported lazily to avoid a configuration/providers_manager import cycle during cleanup.
1632 from airflow.configuration import conf
1634 conf.invalidate_cache()
1636 self._trigger_info_set.clear()
1637 self._notification_info_set.clear()
1638 self._plugins_set.clear()
1639 self._cli_command_functions_set.clear()
1640 self._cli_command_provider_name_set.clear()
1642 self._initialized = False
1643 self._initialization_stack_trace = None