(self)
| 155 | return is_sensitive_command or is_sensitive_hello |
| 156 | |
| 157 | def _redact(self) -> None: |
| 158 | if self._redacted: |
| 159 | return |
| 160 | self._kwargs = {k: v for k, v in self._kwargs.items() if v is not None} |
| 161 | if "durationMS" in self._kwargs and hasattr(self._kwargs["durationMS"], "total_seconds"): |
| 162 | self._kwargs["durationMS"] = self._kwargs["durationMS"].total_seconds() * 1000 |
| 163 | if "serviceId" in self._kwargs: |
| 164 | self._kwargs["serviceId"] = str(self._kwargs["serviceId"]) |
| 165 | document_length = int(os.getenv("MONGOB_LOG_MAX_DOCUMENT_LENGTH", _DEFAULT_DOCUMENT_LENGTH)) |
| 166 | if document_length < 0: |
| 167 | document_length = _DEFAULT_DOCUMENT_LENGTH |
| 168 | is_server_side_error = self._kwargs.pop("isServerSideError", False) |
| 169 | |
| 170 | for doc_name in _DOCUMENT_NAMES: |
| 171 | doc = self._kwargs.get(doc_name) |
| 172 | if doc: |
| 173 | if doc_name == "failure" and is_server_side_error: |
| 174 | doc = {k: v for k, v in doc.items() if k in _REDACTED_FAILURE_FIELDS} |
| 175 | if doc_name != "failure" and self._is_sensitive(doc_name): |
| 176 | doc = json_util.dumps({}) |
| 177 | else: |
| 178 | truncated_doc = _truncate_documents(doc, document_length)[0] |
| 179 | doc = json_util.dumps( |
| 180 | truncated_doc, |
| 181 | json_options=_JSON_OPTIONS, |
| 182 | default=lambda o: o.__repr__(), |
| 183 | ) |
| 184 | if len(doc) > document_length: |
| 185 | doc = ( |
| 186 | doc.encode()[:document_length].decode("unicode-escape", "ignore") |
| 187 | ) + "..." |
| 188 | self._kwargs[doc_name] = doc |
| 189 | self._redacted = True |
no test coverage detected