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
72
73
74
75
76
77
|
"""Common logging package for spamcheck"""
import logging
import sys
from datetime import datetime
from typing import Any, Optional
from pythonjsonlogger import jsonlogger
from vyper import v
def _log_wrapper(level: str) -> Any:
def inner(func: Any) -> Any:
def wrapper(*args, **kwargs) -> Any:
if kwargs.get("extra") is None:
kwargs["extra"] = {}
kwargs["extra"]["time"] = datetime.utcnow().strftime(
"%Y-%m-%dT%H:%M:%S.%fZ"
)
kwargs["extra"]["level"] = level
kwargs["extra"]["service_name"] = "spamcheck"
return func(*args, **kwargs)
return wrapper
return inner
def _exc_info() -> bool:
if sys.exc_info()[0]:
return True
return False
class Logger:
"""Common JSON logger for spamcheck service."""
def __init__(self) -> None:
self.logger = logging.getLogger("spamcheck")
self.logger.propagate = False
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
self.logger.addHandler(handler)
lvl = v.get_string("log_level").upper()
level = logging.getLevelName(lvl)
self.logger.setLevel(level)
@_log_wrapper("FATAL")
def fatal(self, msg: str, extra: Optional[dict] = None):
"""Logs a message with level FATAL on the spamcheck logger."""
self.logger.fatal(msg, exc_info=_exc_info(), extra=extra)
sys.exit(1)
@_log_wrapper("ERROR")
def error(self, msg: str, extra: Optional[dict] = None):
"""Logs a message with level ERROR on the spamcheck logger."""
self.logger.error(msg, exc_info=_exc_info(), extra=extra)
@_log_wrapper("WARNING")
def warning(self, msg: str, extra: Optional[dict] = None):
"""Logs a message with level WARNING on the spamcheck logger."""
self.logger.warning(msg, extra=extra)
@_log_wrapper("INFO")
def info(self, msg: str, extra: Optional[dict] = None):
"""Logs a message with level INFO on the spamcheck logger."""
self.logger.info(msg, extra=extra)
@_log_wrapper("DEBUG")
def debug(self, msg: str, extra: Optional[dict] = None):
"""Logs a message with level DEBUG on the spamcheck logger."""
self.logger.debug(msg, extra=extra)
logger = Logger()
|
