blob: 8646f85d772000c71703a314fada486e7026e859 (
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
|
"""JSON Formatter using [`msgspec`](https://github.com/jcrist/msgspec)"""
### 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("msgspec", throw_error=True)
import msgspec.json # pylint: disable=wrong-import-position,wrong-import-order
### FUNCTIONS
### ============================================================================
def msgspec_default(obj: Any) -> Any:
"""msgspec 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_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 MsgspecFormatter(core.BaseJsonFormatter):
"""JSON formatter using [`msgspec.json.Encoder`](https://jcristharif.com/msgspec/api.html#msgspec.json.Encoder) for encoding."""
def __init__(
self,
*args,
json_default: core.OptionalCallableOrStr = msgspec_default,
**kwargs,
) -> None:
"""
Args:
args: see [BaseJsonFormatter][pythonjsonlogger.core.BaseJsonFormatter]
json_default: a function for encoding non-standard objects
kwargs: see [BaseJsonFormatter][pythonjsonlogger.core.BaseJsonFormatter]
"""
super().__init__(*args, **kwargs)
self.json_default = core.str_to_object(json_default)
self._encoder = msgspec.json.Encoder(enc_hook=self.json_default)
return
def jsonify_log_record(self, log_record: core.LogRecord) -> str:
"""Returns a json string of the log record."""
return self._encoder.encode(log_record).decode("utf8")
|
