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