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# Licensed to the Apache Software Foundation (ASF) under one
2# or more contributor license agreements. See the NOTICE file
3# distributed with this work for additional information
4# regarding copyright ownership. The ASF licenses this file
5# to you under the Apache License, Version 2.0 (the
6# "License"); you may not use this file except in compliance
7# with the License. You may obtain a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing,
12# software distributed under the License is distributed on an
13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14# KIND, either express or implied. See the License for the
15# specific language governing permissions and limitations
16# under the License.
18"""
19Reusable utilities for creating compatibility layers with fallback imports.
21This module provides the core machinery used by sdk.py and standard/* modules
22to handle import fallbacks between Airflow 3.x and 2.x.
23"""
25from __future__ import annotations
27import importlib
30def create_module_getattr(
31 import_map: dict[str, str | tuple[str, ...]],
32 module_map: dict[str, str | tuple[str, ...]] | None = None,
33 rename_map: dict[str, tuple[str, str, str]] | None = None,
34):
35 """
36 Create a __getattr__ function for lazy imports with fallback support.
38 :param import_map: Dictionary mapping attribute names to module paths (single or tuple for fallback)
39 :param module_map: Dictionary mapping module names to module paths (single or tuple for fallback)
40 :param rename_map: Dictionary mapping new names to (new_path, old_path, old_name) tuples
41 :return: A __getattr__ function that can be assigned at module level
42 """
43 module_map = module_map or {}
44 rename_map = rename_map or {}
46 def __getattr__(name: str):
47 # Check renamed imports first
48 if name in rename_map:
49 new_path, old_path, old_name = rename_map[name]
51 rename_error: ImportError | ModuleNotFoundError | AttributeError | None = None
52 # Try new path with new name first (Airflow 3.x)
53 try:
54 module = __import__(new_path, fromlist=[name])
55 return getattr(module, name)
56 except (ImportError, ModuleNotFoundError, AttributeError) as e:
57 rename_error = e
59 # Fall back to old path with old name (Airflow 2.x)
60 try:
61 module = __import__(old_path, fromlist=[old_name])
62 return getattr(module, old_name)
63 except (ImportError, ModuleNotFoundError, AttributeError):
64 if rename_error:
65 raise ImportError(
66 f"Could not import {name!r} from {new_path!r} or {old_name!r} from {old_path!r}"
67 ) from rename_error
68 raise
70 # Check module imports
71 if name in module_map:
72 value = module_map[name]
73 paths = value if isinstance(value, tuple) else (value,)
75 module_error: ImportError | ModuleNotFoundError | None = None
76 for module_path in paths:
77 try:
78 return importlib.import_module(module_path)
79 except (ImportError, ModuleNotFoundError) as e:
80 module_error = e
81 continue
83 if module_error:
84 raise ImportError(f"Could not import module {name!r} from any of: {paths}") from module_error
86 # Check regular imports
87 if name in import_map:
88 value = import_map[name]
89 paths = value if isinstance(value, tuple) else (value,)
91 attr_error: ImportError | ModuleNotFoundError | AttributeError | None = None
92 for module_path in paths:
93 try:
94 module = __import__(module_path, fromlist=[name])
95 return getattr(module, name)
96 except (ImportError, ModuleNotFoundError, AttributeError) as e:
97 attr_error = e
98 continue
100 if attr_error:
101 raise ImportError(f"Could not import {name!r} from any of: {paths}") from attr_error
103 raise AttributeError(f"module has no attribute {name!r}")
105 return __getattr__