Coverage for stlog/kvformatter.py: 91%

79 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-08-21 07:31 +0000

1from __future__ import annotations 

2 

3import json 

4from abc import ABC, abstractmethod 

5from dataclasses import dataclass 

6from typing import Any 

7 

8from stlog.base import check_env_true, logfmt_format_value 

9 

10STLOG_DEFAULT_IGNORE_COMPOUND_TYPES = check_env_true( 

11 "STLOG_IGNORE_COMPOUND_TYPES", True 

12) 

13 

14 

15def _truncate_str(str_value: str, limit: int = 0) -> str: 

16 if limit <= 0: 

17 return str_value 

18 if len(str_value) > limit: 

19 return str_value[0 : (limit - 3)] + "..." 

20 return str_value 

21 

22 

23def _truncate_serialize(value: Any, limit: int = 0) -> str: 

24 try: 

25 serialized = str(value) 

26 except Exception: 

27 serialized = "[can't serialize]" 

28 return _truncate_str(serialized, limit) 

29 

30 

31@dataclass 

32class KVFormatter(ABC): 

33 """Abstract base class to format extras key-values. 

34 

35 Attributes: 

36 value_max_serialized_length: maximum size of extra values to be included in `{extras}` placeholder 

37 (after this limit, the value will be truncated and ... will be added at the end, 0 means "no limit", 

38 default: 40). 

39 

40 """ 

41 

42 value_max_serialized_length: int | None = None 

43 

44 def __post_init__(self): 

45 if self.value_max_serialized_length is None: 

46 self.value_max_serialized_length = 40 

47 

48 def _serialize_value(self, v: Any) -> str: 

49 return _truncate_serialize( 

50 v, 

51 self.value_max_serialized_length 

52 if self.value_max_serialized_length is not None 

53 else 40, 

54 ) 

55 

56 @abstractmethod 

57 def format(self, kvs: dict[str, Any]) -> str: 

58 pass 

59 

60 

61@dataclass 

62class EmptyKVFormatter(KVFormatter): 

63 """Class to format extra key-values as an empty string.""" 

64 

65 def format(self, kvs: dict[str, Any]) -> str: 

66 return "" 

67 

68 

69# Adapted from https://github.com/Mergifyio/daiquiri/blob/main/daiquiri/formatter.py 

70@dataclass 

71class TemplateKVFormatter(KVFormatter): 

72 """Class to format extra key-values as a string with templates. 

73 

74 Example:: 

75 

76 {foo="bar", foo2=123} 

77 

78 will be formatted as: 

79 

80 [foo: bar] [foo2: 123] 

81 

82 Attributes: 

83 template: the template to format a key/value: 

84 `{key}` placeholder is the key, `{value}` is the value 

85 (default to `"{key}={value}"`) 

86 separator: the separator between multiple key/values 

87 prefix: the prefix before key/value parts. 

88 suffix: the suffix after key/values parts. 

89 ignore_compound_types: if set to False, accept compound types (dict, list) as values (they will be 

90 serialized using their default string serialization method) 

91 

92 """ 

93 

94 template: str | None = None 

95 separator: str = ", " 

96 prefix: str = " {" 

97 suffix: str = "}" 

98 ignore_compound_types: bool = STLOG_DEFAULT_IGNORE_COMPOUND_TYPES 

99 

100 def __post_init__(self): 

101 if self.template is None: 

102 self.template = "{key}={value}" 

103 return super().__post_init__() 

104 

105 def format(self, kvs: dict[str, Any]) -> str: 

106 res: str = "" 

107 tmp: list[str] = [] 

108 for k, v in sorted(kvs.items(), key=lambda x: x[0]): 

109 if self.ignore_compound_types and isinstance(v, (dict, list, set)): 

110 continue 

111 assert self.template is not None 

112 tmp.append(self.template.format(key=k, value=self._serialize_value(v))) 

113 res = self.separator.join(tmp) 

114 if res != "": 

115 res = self.prefix + res + self.suffix 

116 return res 

117 

118 

119@dataclass 

120class LogFmtKVFormatter(TemplateKVFormatter): 

121 """Class to format extra key-values as a LogFmt string. 

122 

123 Example:: 

124 

125 {foo="bar", foo2=123, foo3="string with a space"} 

126 

127 will be formatted as: 

128 

129 foo=bar, foo2=123, foo3="string with a space" 

130 

131 

132 Note: `template` and `separator` are automatically set/forced. 

133 

134 """ 

135 

136 def __post_init__(self): 

137 self.separator = " " 

138 if self.template is None: 

139 self.template = "{key}={value}" 

140 if self.value_max_serialized_length is None: 

141 self.value_max_serialized_length = 0 

142 return super().__post_init__() 

143 

144 def _serialize_value(self, v: Any) -> str: 

145 return logfmt_format_value(super()._serialize_value(v)) 

146 

147 

148@dataclass 

149class JsonKVFormatter(KVFormatter): 

150 """Class to format extra key-values as a JSON string. 

151 

152 Attributes: 

153 indent: if set as a positive integer, use this number of spaces to indent the output. 

154 (warning: if you use this KVFormatter 

155 through a JSONFormatter, this parameter can be overriden at the Formatter level). 

156 sort_keys: if True (default), sort keys (warning: if you use this KVFormatter 

157 through a JSONFormatter, this parameter can be overriden at the Formatter level). 

158 """ 

159 

160 indent: int | None = None 

161 sort_keys: bool = True 

162 

163 def __post_init__(self): 

164 if self.value_max_serialized_length is None: 

165 self.value_max_serialized_length = 0 # no limit 

166 self.separator = " " 

167 self.template = "{key}={value}" 

168 return super().__post_init__() 

169 

170 def format(self, kvs: dict[str, Any]) -> str: 

171 return json.dumps( 

172 kvs, 

173 sort_keys=self.sort_keys, 

174 default=self._serialize_value, 

175 indent=self.indent, 

176 )