Skip to content

dp3.api.internal.dp_logger

DPLogger 

DPLogger(config: dict)

Datapoint logger

Logs good/bad datapoints into file for further analysis. They are logged in JSON format. Bad datapoints are logged together with their error message.

Logging may be disabled in api.yml configuration file:

# ...
datapoint_logger:
  good_log: false
  bad_log: false
# ...
Source code in dp3/api/internal/dp_logger.py 
29
30
31
32
33
34
35
36
37
38
def __init__(self, config: dict):
    if not config:
        config = {}

    good_log_file = config.get("good_log", False)
    bad_log_file = config.get("bad_log", False)

    # Setup loggers
    self._good_logger = self.setup_logger("GOOD", good_log_file)
    self._bad_logger = self.setup_logger("BAD", bad_log_file)

setup_logger 

setup_logger(name: str, log_file: str)

Creates new logger instance with log_file as target

Source code in dp3/api/internal/dp_logger.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def setup_logger(self, name: str, log_file: str):
    """Creates new logger instance with `log_file` as target"""
    # Create log handler
    if log_file:
        parent_path = pathlib.Path(log_file).parent
        if not parent_path.exists():
            raise FileNotFoundError(
                f"The directory {parent_path} does not exist,"
                " check the configured path or create the directory."
            )
        log_handler = logging.FileHandler(log_file)
        log_handler.setFormatter(self.LOG_FORMATTER)
    else:
        log_handler = logging.NullHandler()

    # Get logger instance
    logger = logging.getLogger(name)
    logger.addHandler(log_handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False

    return logger

log_good 

log_good(dps: list[DataPointBase], src: str = UNKNOWN_SRC_MSG)

Logs good datapoints

Datapoints are logged one-by-one in processed form. Source should be IP address of incomping request.

Source code in dp3/api/internal/dp_logger.py 
63
64
65
66
67
68
69
70
def log_good(self, dps: list[DataPointBase], src: str = UNKNOWN_SRC_MSG):
    """Logs good datapoints

    Datapoints are logged one-by-one in processed form.
    Source should be IP address of incomping request.
    """
    for dp in dps:
        self._good_logger.info(dp.model_dump(), extra={"src": src})

log_bad 

log_bad(validation_error_msg: str, src: str = UNKNOWN_SRC_MSG)

Logs validation error message, which includes bad input datapoints

Should be called for each individual error (JSON string is expected). Source should be IP address of incoming request.

Source code in dp3/api/internal/dp_logger.py 
72
73
74
75
76
77
78
def log_bad(self, validation_error_msg: str, src: str = UNKNOWN_SRC_MSG):
    """Logs validation error message, which includes bad input datapoints

    Should be called for each individual error (JSON string is expected).
    Source should be IP address of incoming request.
    """
    self._bad_logger.info(validation_error_msg, extra={"src": src})