blob: 16db8426dfa86fe70040a410088e64cebba475da (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
"""JSON Formatter using [orjson](https://github.com/ijl/orjson)"""
### IMPORTS
### ============================================================================
## Future
from __future__ import annotations
## Standard Library
from typing import Any
## Installed
## Application
from . import core
from . import defaults as d
from .utils import package_is_available
# We import msgspec after checking it is available
package_is_available("orjson", throw_error=True)
import orjson # pylint: disable=wrong-import-position,wrong-import-order
### FUNCTIONS
### ============================================================================
def orjson_default(obj: Any) -> Any:
"""orjson default encoder function for non-standard types"""
if d.use_exception_default(obj):
return d.exception_default(obj)
if d.use_traceback_default(obj):
return d.traceback_default(obj)
if d.use_bytes_default(obj):
return d.bytes_default(obj)
if d.use_enum_default(obj):
return d.enum_default(obj)
if d.use_type_default(obj):
return d.type_default(obj)
return d.unknown_default(obj)
### CLASSES
### ============================================================================
class OrjsonFormatter(core.BaseJsonFormatter):
"""JSON formatter using [orjson](https://github.com/ijl/orjson) for encoding."""
def __init__(
self,
*args,
json_default: core.OptionalCallableOrStr = orjson_default,
json_indent: bool = False,
**kwargs,
) -> None:
"""
Args:
args: see [BaseJsonFormatter][pythonjsonlogger.core.BaseJsonFormatter]
json_default: a function for encoding non-standard objects
json_indent: indent output with 2 spaces.
kwargs: see [BaseJsonFormatter][pythonjsonlogger.core.BaseJsonFormatter]
"""
super().__init__(*args, **kwargs)
self.json_default = core.str_to_object(json_default)
self.json_indent = json_indent
return
def jsonify_log_record(self, log_record: core.LogRecord) -> str:
"""Returns a json string of the log record."""
opt = orjson.OPT_NON_STR_KEYS
if self.json_indent:
opt |= orjson.OPT_INDENT_2
return orjson.dumps(log_record, default=self.json_default, option=opt).decode("utf8")
|
