Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/metrics/base_stats_logger.py: 96%

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 Any 

21 

22from airflow.metrics.protocols import DeltaType, Timer, TimerProtocol 

23from airflow.typing_compat import Protocol 

24 

25 

26class StatsLogger(Protocol): 

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

28 

29 @classmethod 

30 def incr( 

31 cls, 

32 stat: str, 

33 count: int = 1, 

34 rate: int | float = 1, 

35 *, 

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

37 ) -> None: 

38 """Increment stat.""" 

39 

40 @classmethod 

41 def decr( 

42 cls, 

43 stat: str, 

44 count: int = 1, 

45 rate: int | float = 1, 

46 *, 

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

48 ) -> None: 

49 """Decrement stat.""" 

50 

51 @classmethod 

52 def gauge( 

53 cls, 

54 stat: str, 

55 value: float, 

56 rate: int | float = 1, 

57 delta: bool = False, 

58 *, 

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

60 ) -> None: 

61 """Gauge stat.""" 

62 

63 @classmethod 

64 def timing( 

65 cls, 

66 stat: str, 

67 dt: DeltaType | None, 

68 *, 

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

70 ) -> None: 

71 """Stats timing.""" 

72 

73 @classmethod 

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

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

76 raise NotImplementedError() 

77 

78 

79class NoStatsLogger: 

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

81 

82 @classmethod 

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

84 """Increment stat.""" 

85 

86 @classmethod 

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

88 """Decrement stat.""" 

89 

90 @classmethod 

91 def gauge( 

92 cls, 

93 stat: str, 

94 value: int, 

95 rate: int = 1, 

96 delta: bool = False, 

97 *, 

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

99 ) -> None: 

100 """Gauge stat.""" 

101 

102 @classmethod 

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

104 """Stats timing.""" 

105 

106 @classmethod 

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

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

109 return Timer()