nolas/logging_config.py at main · levco/nolas · GitHub

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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import logging
import logging.config
from typing import Any

from pythonjsonlogger.json import JsonFormatter

from app.environment import EnvironmentName
from settings import settings

JSON_FORMAT = (
    "%(module)s %(asctime)s %(levelname)s %(thread)d %(processName)s %(task_name)s %(task_id)s %(name)s "
    "%(funcName)s %(filename)s %(lineno)d %(message)s"
)
LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "jsonFormat": {
            "format": JSON_FORMAT,
            "class": "logging_config.CustomJsonFormatter",
        },
        "standardFormat": {
            "format": (
                "%(asctime)s %(levelname)s %(thread)d %(processName)s [%(name)s] [%(funcName)s] "
                "[%(filename)s:%(lineno)d] - %(message)s"
            )
        },
        "standardTaskFormat": {
            "format": (
                "%(asctime)s %(levelname)s %(thread)d %(processName)s %(task_name)s [%(task_id)s] [%(name)s] "
                "[%(funcName)s] [%(filename)s:%(lineno)d] - %(message)s"
            )
        },
    },
    "handlers": {
        "jsonStreamHandler": {
            "formatter": "jsonFormat",
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stdout",  # Default is stderr
        },
    },
    "loggers": {
        "": {"handlers": ["jsonStreamHandler"], "level": settings.logging.level, "propagate": False},
        # werkzeug logging request twice.  This ensures it only logged once (propogate set to false)
        "werkzeug": {
            "handlers": ["jsonStreamHandler"],
            "propagate": False,
        },
    },
}
LOCAL_LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "standard": {"format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s"},
    },
    "handlers": {
        "default": {
            "formatter": "standard",
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stdout",  # Default is stderr
        },
    },
    "loggers": {
        "": {
            "handlers": ["default"],
            "level": settings.logging.level,
            "propagate": False,
        },
        # werkzeug logging request twice.  This ensures it only logged once (propogate set to false)
        "werkzeug": {"handlers": ["default"], "propagate": False},
        "gql": {"handlers": ["default"], "level": logging.WARNING, "propagate": False},
        "botocore": {"handlers": ["default"], "level": logging.WARNING, "propagate": False},
        "urllib3": {"handlers": ["default"], "level": logging.WARNING, "propagate": False},
        "pymongo": {"handlers": ["default"], "level": logging.WARNING, "propagate": False},
        "ddtrace": {"handlers": ["default"], "level": logging.WARNING, "propagate": False},
    },
}


# Used because we add custom local formatting.
class CustomJsonFormatter(JsonFormatter):
    _app_env: str | None
    _pretty_format: bool

    def __init__(self, *args: Any, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self._app_env = settings.environment
        self._pretty_format = settings.logging.use_pretty_json
        # For local development we want to make logs more clear
        if self._app_env == EnvironmentName.DEVELOPMENT.value and self._pretty_format:
            self.json_indent = 2

    def format(self, record: logging.LogRecord) -> str:
        result = super(CustomJsonFormatter, self).format(record)
        # For local development we want to make logs more clear.
        if self._app_env == EnvironmentName.DEVELOPMENT.value and self._pretty_format:
            result = result.replace("\\n", "\n\t\t")
        return result


def setup_logging() -> None:
    if settings.logging.use_config is True:
        """Setup root logger using our logging config"""
        logging.config.dictConfig(LOGGING_CONFIG)
        logging.captureWarnings(True)
        logging.disable(logging.NOTSET)
    else:
        logging.config.dictConfig(LOCAL_LOGGING_CONFIG)
        logging.captureWarnings(True)
        logging.disable(logging.NOTSET)