Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/utils/deprecation_tools.py: 76%

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. 

17from __future__ import annotations 

18 

19import functools 

20import importlib 

21import sys 

22import warnings 

23from types import ModuleType 

24 

25 

26def getattr_with_deprecation( 

27 imports: dict[str, str], 

28 module: str, 

29 override_deprecated_classes: dict[str, str], 

30 extra_message: str, 

31 name: str, 

32): 

33 """ 

34 Retrieve the imported attribute from the redirected module and raises a deprecation warning. 

35 

36 :param imports: dict of imports and their redirection for the module 

37 :param module: name of the module in the package to get the attribute from 

38 :param override_deprecated_classes: override target classes with deprecated ones. If target class is 

39 found in the dictionary, it will be displayed in the warning message. 

40 :param extra_message: extra message to display in the warning or import error message 

41 :param name: attribute name 

42 :return: 

43 """ 

44 target_class_full_name = imports.get(name) 

45 if not target_class_full_name: 

46 raise AttributeError(f"The module `{module!r}` has no attribute `{name!r}`") 

47 warning_class_name = target_class_full_name 

48 if override_deprecated_classes and name in override_deprecated_classes: 

49 warning_class_name = override_deprecated_classes[name] 

50 message = f"The `{module}.{name}` class is deprecated. Please use `{warning_class_name!r}`." 

51 if extra_message: 

52 message += f" {extra_message}." 

53 warnings.warn(message, DeprecationWarning, stacklevel=2) 

54 new_module, new_class_name = target_class_full_name.rsplit(".", 1) 

55 try: 

56 return getattr(importlib.import_module(new_module), new_class_name) 

57 except ImportError as e: 

58 error_message = ( 

59 f"Could not import `{new_module}.{new_class_name}` while trying to import `{module}.{name}`." 

60 ) 

61 if extra_message: 

62 error_message += f" {extra_message}." 

63 raise ImportError(error_message) from e 

64 

65 

66def add_deprecated_classes( 

67 module_imports: dict[str, dict[str, str]], 

68 package: str, 

69 override_deprecated_classes: dict[str, dict[str, str]] | None = None, 

70 extra_message: str | None = None, 

71): 

72 """ 

73 Add deprecated class PEP-563 imports and warnings modules to the package. 

74 

75 :param module_imports: imports to use 

76 :param package: package name 

77 :param override_deprecated_classes: override target classes with deprecated ones. If module + 

78 target class is found in the dictionary, it will be displayed in the warning message. 

79 :param extra_message: extra message to display in the warning or import error message 

80 """ 

81 for module_name, imports in module_imports.items(): 

82 full_module_name = f"{package}.{module_name}" 

83 module_type = ModuleType(full_module_name) 

84 if override_deprecated_classes and module_name in override_deprecated_classes: 

85 override_deprecated_classes_for_module = override_deprecated_classes[module_name] 

86 else: 

87 override_deprecated_classes_for_module = {} 

88 

89 # Mypy is not able to derive the right function signature https://github.com/python/mypy/issues/2427 

90 module_type.__getattr__ = functools.partial( # type: ignore[assignment] 

91 getattr_with_deprecation, 

92 imports, 

93 full_module_name, 

94 override_deprecated_classes_for_module, 

95 extra_message or "", 

96 ) 

97 sys.modules.setdefault(full_module_name, module_type)