Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/airflow/sdk/definitions/_internal/logging_mixin.py: 58%

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 

19from typing import TYPE_CHECKING, Any, TypeVar 

20 

21import structlog 

22 

23if TYPE_CHECKING: 

24 from airflow.sdk.types import Logger 

25 

26_T = TypeVar("_T") 

27 

28 

29class LoggingMixin: 

30 """Convenience super-class to have a logger configured with the class name.""" 

31 

32 _log: Logger | None = None 

33 

34 # Parent logger used by this class. It should match one of the loggers defined in the 

35 # `logging_config_class`. By default, this attribute is used to create the final name of the logger, and 

36 # will prefix the `_logger_name` with a separating dot. 

37 _log_config_logger_name: str | None = None 

38 

39 _logger_name: str | None = None 

40 

41 def __init__(self, context=None): 

42 self._set_context(context) 

43 super().__init__() 

44 

45 @staticmethod 

46 def _create_logger_name( 

47 logged_class: type[_T], 

48 log_config_logger_name: str | None = None, 

49 class_logger_name: str | None = None, 

50 ) -> str: 

51 """ 

52 Generate a logger name for the given `logged_class`. 

53 

54 By default, this function returns the `class_logger_name` as logger name. If it is not provided, 

55 the {class.__module__}.{class.__name__} is returned instead. When a `parent_logger_name` is provided, 

56 it will prefix the logger name with a separating dot. 

57 """ 

58 logger_name: str = ( 

59 class_logger_name 

60 if class_logger_name is not None 

61 else f"{logged_class.__module__}.{logged_class.__name__}" 

62 ) 

63 

64 if log_config_logger_name: 

65 return f"{log_config_logger_name}.{logger_name}" if logger_name else log_config_logger_name 

66 return logger_name 

67 

68 @classmethod 

69 def _get_log(cls, obj: Any, clazz: type[_T]) -> Logger: 

70 if obj._log is None: 

71 logger_name: str = cls._create_logger_name( 

72 logged_class=clazz, 

73 log_config_logger_name=obj._log_config_logger_name, 

74 class_logger_name=obj._logger_name, 

75 ) 

76 obj._log = structlog.get_logger(logger_name) 

77 return obj._log 

78 

79 @classmethod 

80 def logger(cls) -> Logger: 

81 """Return a logger.""" 

82 return LoggingMixin._get_log(cls, cls) 

83 

84 @property 

85 def log(self) -> Logger: 

86 """Return a logger.""" 

87 return LoggingMixin._get_log(self, self.__class__) 

88 

89 def _set_context(self, context): ...