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

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(imports: dict[str, str], module: str, name: str): 

27 target_class_full_name = imports.get(name) 

28 if not target_class_full_name: 

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

30 warnings.warn( 

31 f"The `{module}.{name}` class is deprecated. Please use `{target_class_full_name!r}`.", 

32 DeprecationWarning, 

33 stacklevel=2, 

34 ) 

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

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

37 

38 

39def add_deprecated_classes(module_imports: dict[str, dict[str, str]], package: str): 

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

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

42 module_type = ModuleType(full_module_name) 

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

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

45 getattr_with_deprecation, imports, full_module_name 

46 ) 

47 sys.modules.setdefault(full_module_name, module_type)