Add opt-in HTTP diagnostics logging with automatic header redaction

src/PowerPlatform/Dataverse/core/_http.py

@@ -12,10 +12,13 @@
 from __future__ import annotations
 
 import time
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 import requests
 
+if TYPE_CHECKING:
+    from ._http_logger import _HttpLogger
+
 
 class _HttpClient:
     """
@@ -34,6 +37,9 @@ class _HttpClient:
         When provided, all requests use this session (enabling TCP/TLS reuse).
         When ``None``, each request uses ``requests.request()`` directly.
     :type session: :class:`requests.Session` | None
+    :param logger: Optional HTTP diagnostics logger. When provided, all requests,
+        responses, and transport errors are logged with automatic header redaction.
+    :type logger: ~PowerPlatform.Dataverse.core._http_logger._HttpLogger | None
     """
 
     def __init__(
@@ -42,11 +48,13 @@ def __init__(
         backoff: Optional[float] = None,
         timeout: Optional[float] = None,
         session: Optional[requests.Session] = None,
+        logger: Optional["_HttpLogger"] = None,
     ) -> None:
         self.max_attempts = retries if retries is not None else 5
         self.base_delay = backoff if backoff is not None else 0.5
         self.default_timeout: Optional[float] = timeout
         self._session = session
+        self._logger = logger
 
     def _request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
         """
@@ -73,12 +81,36 @@ def _request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
                 m = (method or "").lower()
                 kwargs["timeout"] = 120 if m in ("post", "delete") else 10
 
+        # Log outbound request once (before retry loop)
+        if self._logger is not None:
+            self._logger.log_request(
+                method,
+                url,
+                headers=kwargs.get("headers"),
+                body=kwargs.get("json") or kwargs.get("data"),
+            )
+
         # Small backoff retry on network errors only
         requester = self._session.request if self._session is not None else requests.request
         for attempt in range(self.max_attempts):
             try:
-                return requester(method, url, **kwargs)
-            except requests.exceptions.RequestException:
+                t0 = time.monotonic()
+                resp = requester(method, url, **kwargs)
+                elapsed_ms = (time.monotonic() - t0) * 1000
+
+                if self._logger is not None:
+                    self._logger.log_response(
+                        method,
+                        url,
+                        status_code=resp.status_code,
+                        headers=dict(resp.headers),
+                        body=resp.text,
+                        elapsed_ms=elapsed_ms,
+                    )
+                return resp
+            except requests.exceptions.RequestException as exc:
+                if self._logger is not None:
+                    self._logger.log_error(method, url, exc)
                 if attempt == self.max_attempts - 1:
                     raise
                 delay = self.base_delay * (2**attempt)

src/PowerPlatform/Dataverse/core/_http_logger.py

@@ -0,0 +1,119 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+Internal HTTP logger that writes redacted request/response diagnostics to local files.
+"""
+
+from __future__ import annotations
+
+import json as _json
+import logging
+import os
+import uuid
+from datetime import datetime, timezone
+from logging.handlers import RotatingFileHandler
+from typing import Any, Dict, Optional
+
+from .log_config import LogConfig
+
+
+class _HttpLogger:
+    """Structured HTTP diagnostic logger with automatic header redaction."""
+
+    def __init__(self, config: LogConfig) -> None:
+        self._config = config
+        self._redacted = {h.lower() for h in config.redacted_headers}
+
+        # Ensure folder exists
+        os.makedirs(config.log_folder, exist_ok=True)
+
+        # Build timestamped filename
+        ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
+        filename = f"{config.log_file_prefix}_{ts}.log"
+        filepath = os.path.join(config.log_folder, filename)
+
+        # Create a dedicated named logger (not root) to avoid side effects
+        logger_name = f"PowerPlatform.Dataverse.http.{uuid.uuid4().hex[:8]}"
+        self._logger = logging.getLogger(logger_name)
+        self._logger.setLevel(getattr(logging, config.log_level.upper(), logging.DEBUG))
+        self._logger.propagate = False  # don't bubble to root
+
+        handler = RotatingFileHandler(
+            filepath,
+            maxBytes=config.max_file_bytes,
+            backupCount=config.backup_count,
+            encoding="utf-8",
+        )
+        formatter = logging.Formatter(
+            "[%(asctime)s] %(levelname)s %(message)s",
+            datefmt="%Y-%m-%dT%H:%M:%S%z",
+        )
+        handler.setFormatter(formatter)
+        self._logger.addHandler(handler)
+
+    def log_request(
+        self,
+        method: str,
+        url: str,
+        headers: Optional[Dict[str, str]] = None,
+        body: Any = None,
+    ) -> None:
+        """Log an outbound HTTP request."""
+        safe_headers = self._redact_headers(headers or {})
+        body_text = self._truncate_body(body)
+        lines = [
+            f">>> REQUEST  {method.upper()} {url}",
+            f"    Headers: {safe_headers}",
+        ]
+        if body_text:
+            lines.append(f"    Body:    {body_text}")
+        self._logger.debug("\n".join(lines))
+
+    def log_response(
+        self,
+        method: str,
+        url: str,
+        status_code: int,
+        headers: Optional[Dict[str, str]] = None,
+        body: Any = None,
+        elapsed_ms: Optional[float] = None,
+    ) -> None:
+        """Log an inbound HTTP response."""
+        safe_headers = self._redact_headers(headers or {})
+        body_text = self._truncate_body(body)
+        elapsed_str = f" ({elapsed_ms:.1f}ms)" if elapsed_ms is not None else ""
+        lines = [
+            f"<<< RESPONSE {status_code} {method.upper()} {url}{elapsed_str}",
+            f"    Headers: {safe_headers}",
+        ]
+        if body_text:
+            lines.append(f"    Body:    {body_text}")
+        self._logger.debug("\n".join(lines))
+
+    def log_error(self, method: str, url: str, error: Exception) -> None:
+        """Log an HTTP transport error."""
+        self._logger.error(f"!!! ERROR    {method.upper()} {url} - {type(error).__name__}: {error}")
+
+    def _redact_headers(self, headers: Dict[str, str]) -> Dict[str, str]:
+        return {k: ("[REDACTED]" if k.lower() in self._redacted else v) for k, v in headers.items()}
+
+    def _truncate_body(self, body: Any) -> str:
+        if body is None:
+            return ""
+        if isinstance(body, (bytes, bytearray)):
+            text = body.decode("utf-8", errors="replace")
+        elif not isinstance(body, str):
+            try:
+                text = _json.dumps(body, default=str, ensure_ascii=False)
+            except (TypeError, ValueError):
+                text = str(body)
+        else:
+            text = body
+
+        limit = self._config.max_body_bytes
+        if limit == 0:
+            return ""
+        if len(text) > limit:
+            return text[:limit] + f"... [truncated, {len(text)} bytes total]"
+        return text

src/PowerPlatform/Dataverse/core/config.py

@@ -12,7 +12,10 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
+
+if TYPE_CHECKING:
+    from .log_config import LogConfig
 
 
 @dataclass(frozen=True)
@@ -28,6 +31,10 @@ class DataverseConfig:
     :type http_backoff: :class:`float` or None
     :param http_timeout: Optional request timeout in seconds. Reserved for future use.
     :type http_timeout: :class:`float` or None
+    :param log_config: Optional local HTTP diagnostics logging configuration.
+        When provided, all HTTP requests and responses are logged to timestamped
+        ``.log`` files with automatic redaction of sensitive headers.
+    :type log_config: ~PowerPlatform.Dataverse.core.log_config.LogConfig or None
     """
 
     language_code: int = 1033
@@ -37,6 +44,8 @@ class DataverseConfig:
     http_backoff: Optional[float] = None
     http_timeout: Optional[float] = None
 
+    log_config: Optional["LogConfig"] = None
+
     @classmethod
     def from_env(cls) -> "DataverseConfig":
         """
@@ -51,4 +60,5 @@ def from_env(cls) -> "DataverseConfig":
             http_retries=None,
             http_backoff=None,
             http_timeout=None,
+            log_config=None,
         )

src/PowerPlatform/Dataverse/core/log_config.py

@@ -0,0 +1,64 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+Local file logging configuration for Dataverse SDK HTTP diagnostics.
+
+Provides :class:`LogConfig`, an opt-in configuration for writing request/response
+traces to ``.log`` files with automatic header redaction and timestamped filenames.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import FrozenSet
+
+__all__ = ["LogConfig"]
+
+# Headers whose values must never appear in log files
+_DEFAULT_REDACTED_HEADERS: FrozenSet[str] = frozenset(
+    {
+        "authorization",
+        "proxy-authorization",
+        "x-ms-authorization-auxiliary",
+        "ocp-apim-subscription-key",
+    }
+)
+
+
+def _default_redacted_headers() -> FrozenSet[str]:
+    return _DEFAULT_REDACTED_HEADERS
+
+
+@dataclass(frozen=True)
+class LogConfig:
+    """
+    Configuration for local HTTP diagnostics logging.
+
+    When provided to :class:`~PowerPlatform.Dataverse.client.DataverseClient` via
+    :class:`~PowerPlatform.Dataverse.core.config.DataverseConfig`, every HTTP request
+    and response is logged to timestamped ``.log`` files in the specified folder.
+    Sensitive headers (e.g. ``Authorization``) are automatically redacted.
+
+    :param log_folder: Directory path for log files. Created automatically if missing.
+        Default: ``"./dataverse_logs"``
+    :param log_file_prefix: Filename prefix. Timestamp is appended automatically.
+        Default: ``"dataverse"``  →  ``dataverse_20260310_143022.log``
+    :param max_body_bytes: Maximum bytes of request/response body to capture.
+        ``0`` disables body logging. Default: ``4096``.
+    :param redacted_headers: Header names (case-insensitive) whose values are
+        replaced with ``"[REDACTED]"`` in logs. Defaults include
+        ``Authorization``, ``Proxy-Authorization``, etc.
+    :param log_level: Python logging level name. Default: ``"DEBUG"``.
+    :param max_file_bytes: Max size per log file before rotation (bytes).
+        Default: ``10_485_760`` (10 MB).
+    :param backup_count: Number of rotated backup files to keep. Default: ``5``.
+    """
+
+    log_folder: str = "./dataverse_logs"
+    log_file_prefix: str = "dataverse"
+    max_body_bytes: int = 4096
+    redacted_headers: FrozenSet[str] = field(default_factory=_default_redacted_headers)
+    log_level: str = "DEBUG"
+    max_file_bytes: int = 10_485_760  # 10 MB
+    backup_count: int = 5

src/PowerPlatform/Dataverse/data/_odata.py

@@ -143,13 +143,18 @@ def __init__(
                 "PowerPlatform.Dataverse.core.config", fromlist=["DataverseConfig"]
             ).DataverseConfig.from_env()
         )
+        _log = None
+        if self.config.log_config is not None:
+            from ..core._http_logger import _HttpLogger
+
+            _log = _HttpLogger(self.config.log_config)
         self._http = _HttpClient(
             retries=self.config.http_retries,
             backoff=self.config.http_backoff,
             timeout=self.config.http_timeout,
             session=session,
+            logger=_log,
         )
-        # Cache: normalized table_schema_name (lowercase) -> entity set name (plural) resolved from metadata
         self._logical_to_entityset_cache: dict[str, str] = {}
         # Cache: normalized table_schema_name (lowercase) -> primary id attribute (e.g. accountid)
         self._logical_primaryid_cache: dict[str, str] = {}