Class LogMessage

pymongo/logger.py:127–189

class LogMessage:
    __slots__ = ("_kwargs", "_redacted")

    def __init__(self, **kwargs: Any):
        self._kwargs = kwargs
        self._redacted = False

    def __str__(self) -> str:
        self._redact()
        return "%s" % (
            json_util.dumps(
                self._kwargs, json_options=_JSON_OPTIONS, default=lambda o: o.__repr__()
            )
        )

    def _is_sensitive(self, doc_name: str) -> bool:
        is_speculative_authenticate = (
            self._kwargs.pop("speculative_authenticate", False)
            or "speculativeAuthenticate" in self._kwargs[doc_name]
        )
        is_sensitive_command = (
            "commandName" in self._kwargs and self._kwargs["commandName"] in _SENSITIVE_COMMANDS
        )

        is_sensitive_hello = (
            self._kwargs.get("commandName", None) in _HELLO_COMMANDS and is_speculative_authenticate
        )

        return is_sensitive_command or is_sensitive_hello

    def _redact(self) -> None:
        if self._redacted:
            return
        self._kwargs = {k: v for k, v in self._kwargs.items() if v is not None}
        if "durationMS" in self._kwargs and hasattr(self._kwargs["durationMS"], "total_seconds"):
            self._kwargs["durationMS"] = self._kwargs["durationMS"].total_seconds() * 1000
        if "serviceId" in self._kwargs:
            self._kwargs["serviceId"] = str(self._kwargs["serviceId"])
        document_length = int(os.getenv("MONGOB_LOG_MAX_DOCUMENT_LENGTH", _DEFAULT_DOCUMENT_LENGTH))
        if document_length < 0:
            document_length = _DEFAULT_DOCUMENT_LENGTH
        is_server_side_error = self._kwargs.pop("isServerSideError", False)

        for doc_name in _DOCUMENT_NAMES:
            doc = self._kwargs.get(doc_name)
            if doc:
                if doc_name == "failure" and is_server_side_error:
                    doc = {k: v for k, v in doc.items() if k in _REDACTED_FAILURE_FIELDS}
                if doc_name != "failure" and self._is_sensitive(doc_name):
                    doc = json_util.dumps({})
                else:
                    truncated_doc = _truncate_documents(doc, document_length)[0]
                    doc = json_util.dumps(
                        truncated_doc,
                        json_options=_JSON_OPTIONS,
                        default=lambda o: o.__repr__(),
                    )
                if len(doc) > document_length: