Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/airflow/sdk/_shared/observability/metrics/base_stats_logger.py: 89%

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. 

17 

18from __future__ import annotations 

19 

20from typing import TYPE_CHECKING, Any, Protocol 

21 

22from .protocols import Timer 

23 

24if TYPE_CHECKING: 

25 from .protocols import DeltaType 

26 

27 

28class StatsLogger(Protocol): 

29 """This class is only used for TypeChecking (for IDEs, mypy, etc).""" 

30 

31 instance: StatsLogger | NoStatsLogger | None = None 

32 

33 @classmethod 

34 def initialize( 

35 cls, 

36 *, 

37 is_statsd_datadog_enabled: bool, 

38 is_statsd_on: bool, 

39 is_otel_on: bool, 

40 reset_instance: bool = True, 

41 ) -> None: 

42 """Initialize the StatsLogger instance.""" 

43 

44 @classmethod 

45 def incr( 

46 cls, 

47 stat: str, 

48 count: int = 1, 

49 rate: int | float = 1, 

50 *, 

51 tags: dict[str, Any] | None = None, 

52 ) -> None: 

53 """Increment stat.""" 

54 

55 @classmethod 

56 def decr( 

57 cls, 

58 stat: str, 

59 count: int = 1, 

60 rate: int | float = 1, 

61 *, 

62 tags: dict[str, Any] | None = None, 

63 ) -> None: 

64 """Decrement stat.""" 

65 

66 @classmethod 

67 def gauge( 

68 cls, 

69 stat: str, 

70 value: float, 

71 rate: int | float = 1, 

72 delta: bool = False, 

73 *, 

74 tags: dict[str, Any] | None = None, 

75 ) -> None: 

76 """Gauge stat.""" 

77 

78 @classmethod 

79 def timing( 

80 cls, 

81 stat: str, 

82 dt: DeltaType | None, 

83 *, 

84 tags: dict[str, Any] | None = None, 

85 ) -> None: 

86 """Stats timing.""" 

87 

88 @classmethod 

89 def timer(cls, *args, **kwargs) -> Timer: 

90 """Timer metric that can be cancelled.""" 

91 raise NotImplementedError() 

92 

93 

94class NoStatsLogger: 

95 """If no StatsLogger is configured, NoStatsLogger is used as a fallback.""" 

96 

97 @classmethod 

98 def initialize( 

99 cls, 

100 *, 

101 is_statsd_datadog_enabled: bool, 

102 is_statsd_on: bool, 

103 is_otel_on: bool, 

104 reset_instance: bool = True, 

105 ) -> None: 

106 """Initialize the NoStatsLogger instance.""" 

107 

108 @classmethod 

109 def incr(cls, stat: str, count: int = 1, rate: int = 1, *, tags: dict[str, str] | None = None) -> None: 

110 """Increment stat.""" 

111 

112 @classmethod 

113 def decr(cls, stat: str, count: int = 1, rate: int = 1, *, tags: dict[str, str] | None = None) -> None: 

114 """Decrement stat.""" 

115 

116 @classmethod 

117 def gauge( 

118 cls, 

119 stat: str, 

120 value: int, 

121 rate: int = 1, 

122 delta: bool = False, 

123 *, 

124 tags: dict[str, str] | None = None, 

125 ) -> None: 

126 """Gauge stat.""" 

127 

128 @classmethod 

129 def timing(cls, stat: str, dt: DeltaType, *, tags: dict[str, str] | None = None) -> None: 

130 """Stats timing.""" 

131 

132 @classmethod 

133 def timer(cls, *args, **kwargs) -> Timer: 

134 """Timer metric that can be cancelled.""" 

135 return Timer()