Skip to content
Back to top

# Source code for aiokem.message_logger

    """Module to log messages to files."""
    # pylint: disable=line-too-long

    from __future__ import annotations

    import copy
    import json
    import logging
    from typing import Any

    REDACTED: str = "**redacted**"

    _LOGGER = logging.getLogger(__name__)

    REDACTED_FIELDS = [
        "lat",
        "long",
        "address1",
        "address2",
        "city",
        "state",
        "country",
        "postalCode",
        "serialNumber",
        "deviceIpAddress",
        "macAddress",
        "businessPartnerNo",
        "e164PhoneNumber",
        "displayPhoneNumber",
        "adminEmails",
        "associatedUsers",
        "access_token",
        "refresh_token",
        "id_token",
        "email",
        "firstName",
        "lastName",
    ]
    REDACTED_LISTS = [
        "deviceSerialNumbers",
    ]

    def redact_fields(log_message: Any) -> Any:
        """Removes redacted fields from messages."""
        # This can be optimized, but for now, it works and has a test.
        if isinstance(log_message, dict):
            for k, v in log_message.items():
                if isinstance(v, dict):
                    log_message[k] = redact_fields(v)
                elif isinstance(v, list):
                    if k in REDACTED_LISTS:
                        log_message[k] = [REDACTED for _ in v]
                    else:
                        log_message[k] = [redact_fields(i) for i in v]
            for field in REDACTED_FIELDS:
                if field in log_message:
                    log_message[field] = REDACTED
        elif isinstance(log_message, list):
            for i in range(len(log_message)):
                if isinstance(log_message[i], dict):
                    log_message[i] = redact_fields(log_message[i])
                elif isinstance(log_message[i], list):
                    log_message[i] = [redact_fields(j) for j in log_message[i]]
                for field in REDACTED_FIELDS:
                    if field in log_message[i]:
                        log_message[i][field] = REDACTED
        return log_message

    def log_json_message(json_message: dict[str, Any] | list[dict[str, Any]]) -> None:
        """Logs a JSON message redacting sensitive information."""
        if not json_message:
            _LOGGER.debug("No JSON message to log")
            return
        log_message = redact_fields(copy.deepcopy(json_message))
        _LOGGER.debug(json.dumps(log_message, indent=4))