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

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 incr( 

35 cls, 

36 stat: str, 

37 count: int = 1, 

38 rate: int | float = 1, 

39 *, 

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

41 ) -> None: 

42 """Increment stat.""" 

43 

44 @classmethod 

45 def decr( 

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 """Decrement stat.""" 

54 

55 @classmethod 

56 def gauge( 

57 cls, 

58 stat: str, 

59 value: float, 

60 rate: int | float = 1, 

61 delta: bool = False, 

62 *, 

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

64 ) -> None: 

65 """Gauge stat.""" 

66 

67 @classmethod 

68 def timing( 

69 cls, 

70 stat: str, 

71 dt: DeltaType | None, 

72 *, 

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

74 ) -> None: 

75 """Stats timing.""" 

76 

77 @classmethod 

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

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

80 raise NotImplementedError() 

81 

82 

83class NoStatsLogger: 

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

85 

86 @classmethod 

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

88 """Increment stat.""" 

89 

90 @classmethod 

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

92 """Decrement stat.""" 

93 

94 @classmethod 

95 def gauge( 

96 cls, 

97 stat: str, 

98 value: int, 

99 rate: int = 1, 

100 delta: bool = False, 

101 *, 

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

103 ) -> None: 

104 """Gauge stat.""" 

105 

106 @classmethod 

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

108 """Stats timing.""" 

109 

110 @classmethod 

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

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

113 return Timer()