Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/airflow/metrics/base_stats_logger.py: 94%

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 

21 

22from airflow.metrics.protocols import Timer 

23from airflow.typing_compat import Protocol 

24 

25if TYPE_CHECKING: 

26 from airflow.metrics.protocols import DeltaType, TimerProtocol 

27 

28 

29class StatsLogger(Protocol): 

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

31 

32 instance: StatsLogger | NoStatsLogger | None = None 

33 

34 @classmethod 

35 def incr( 

36 cls, 

37 stat: str, 

38 count: int = 1, 

39 rate: int | float = 1, 

40 *, 

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

42 ) -> None: 

43 """Increment stat.""" 

44 

45 @classmethod 

46 def decr( 

47 cls, 

48 stat: str, 

49 count: int = 1, 

50 rate: int | float = 1, 

51 *, 

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

53 ) -> None: 

54 """Decrement stat.""" 

55 

56 @classmethod 

57 def gauge( 

58 cls, 

59 stat: str, 

60 value: float, 

61 rate: int | float = 1, 

62 delta: bool = False, 

63 *, 

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

65 ) -> None: 

66 """Gauge stat.""" 

67 

68 @classmethod 

69 def timing( 

70 cls, 

71 stat: str, 

72 dt: DeltaType | None, 

73 *, 

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

75 ) -> None: 

76 """Stats timing.""" 

77 

78 @classmethod 

79 def timer(cls, *args, **kwargs) -> TimerProtocol: 

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

81 raise NotImplementedError() 

82 

83 

84class NoStatsLogger: 

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

86 

87 @classmethod 

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

89 """Increment stat.""" 

90 

91 @classmethod 

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

93 """Decrement stat.""" 

94 

95 @classmethod 

96 def gauge( 

97 cls, 

98 stat: str, 

99 value: int, 

100 rate: int = 1, 

101 delta: bool = False, 

102 *, 

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

104 ) -> None: 

105 """Gauge stat.""" 

106 

107 @classmethod 

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

109 """Stats timing.""" 

110 

111 @classmethod 

112 def timer(cls, *args, **kwargs) -> TimerProtocol: 

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

114 return Timer()