|
1 | 1 | import logging |
2 | 2 | import numbers |
| 3 | +import shlex |
3 | 4 | import traceback |
4 | 5 | from types import TracebackType |
5 | | -from typing import Any, Dict, List, Optional, Tuple, Type, cast |
| 6 | +from typing import Any, Dict, List, Optional, Tuple, Type, Union, cast |
6 | 7 |
|
7 | 8 | ExcInfo = Tuple[Type[BaseException], BaseException, TracebackType] |
8 | 9 |
|
|
37 | 38 | ) |
38 | 39 |
|
39 | 40 |
|
| 41 | +def parse_logfmt( |
| 42 | + line: str, aliases: Optional[Dict[str, str]] = None, convert_numeric: bool = False |
| 43 | +) -> Dict[str, Union[str, int, float]]: |
| 44 | + """ |
| 45 | + Parse a logfmt formatted string. |
| 46 | + """ |
| 47 | + aliases = aliases or {} |
| 48 | + fields = {} |
| 49 | + for token in shlex.split(line): |
| 50 | + key, _, value = token.partition("=") |
| 51 | + key = aliases.get(key, key) |
| 52 | + value: Union[str, int, float] = value.replace("\\n", "\n") |
| 53 | + if convert_numeric and value.isdigit(): |
| 54 | + value = int(value) |
| 55 | + elif ( |
| 56 | + convert_numeric |
| 57 | + and value.count(".") == 1 |
| 58 | + and value.replace(".", "").isdigit() |
| 59 | + ): |
| 60 | + value = float(value) |
| 61 | + fields[key] = value |
| 62 | + return fields |
| 63 | + |
| 64 | + |
40 | 65 | class _DefaultFormatter(logging.Formatter): |
41 | 66 | def format(self, record): |
42 | 67 | exc_info = record.exc_info |
|
0 commit comments