chore: add structured logging with secret redaction

CHANGELOG.md

@@ -45,7 +45,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     - `request_timeout: float` (seconds): default `30.0` (was `6.0`; see Behavior changes)
 
   These tune the underlying `httpx.Limits` and `httpx.Timeout`. The existing `http_client=` and `transport=` kwargs continue to act as escape hatches; when `http_client` is set, none of the four new kwargs apply. Env-var fallbacks for the new kwargs: `STREAM_MAX_CONNS_PER_HOST`, `STREAM_IDLE_TIMEOUT`, `STREAM_CONNECT_TIMEOUT`, `STREAM_REQUEST_TIMEOUT`.
-- INFO log on client construction (logger `getstream`) lists the effective pool config and whether a user-supplied `http_client` is in use.
+
+- Structured logging ([CHA-2957](https://linear.app/stream/issue/CHA-2957)).
+  New `logger: logging.Logger | None` and `log_bodies: bool = False` kwargs on
+  `Stream(...)` and `AsyncStream(...)`. Off by default: nothing is emitted
+  unless a logger is passed. Four events, each carrying structured fields via
+  the stdlib logging `extra={}` mechanism:
+    - `client.initialized` (INFO, once at construction): SDK name/version,
+      resolved pool knobs, `gzip_enabled`, `user_http_client`, `log_bodies`.
+      Replaces the old plain-text pool-config INFO line.
+    - `http.request.sent` (DEBUG, before each request): method, path,
+      redacted query, `stream.endpoint_name`.
+    - `http.response.received` (DEBUG, after any response including
+      4xx/5xx): status code, response size, `duration_ms`. HTTP error
+      status codes are data, not failures.
+    - `http.request.failed` (ERROR, transport failure only, no HTTP
+      response received): `error.type` (`connection_reset` / `timeout` /
+      `dns_failure` / `tls_handshake_failed` / `unknown`), `error.message`.
+  Query values for `api_key`/`api_secret`/`token` and top-level JSON body
+  keys `api_secret`/`token`/`password` are always redacted; no opt-out.
+  Bodies are never logged by default; `log_bodies=True` adds redacted
+  request/response bodies and emits one WARNING at construction.
 
 - Webhook handling spec helpers (CHA-2961): `UnknownEvent` dataclass for
   forward-compat; `gunzip_payload`, `decode_sqs_payload`, `decode_sns_payload`

README.md

@@ -99,6 +99,19 @@ response: StreamResponse[StartClosedCaptionsResponse] = call.start_closed_captio
 response.data  # Gives the StartClosedCaptionsResponse model
 ```
 
+### Logging
+
+The SDK emits structured log events (`client.initialized`, `http.request.sent`, `http.response.received`, `http.request.failed`) through the stdlib `logging` module. By default nothing is printed: pass a `logging.Logger` to see them.
+
+```python
+import logging
+
+logging.basicConfig(level=logging.DEBUG)
+client = Stream(api_key="key", api_secret="secret", logger=logging.getLogger("myapp.stream"))
+```
+
+Each event carries structured fields via the standard `extra={}` mechanism (for example `http.response.status_code`, `duration_ms`, `stream.endpoint_name`). Query and body values for known secret keys (`api_key`, `api_secret`, `token`, `password`) are always redacted. Request/response bodies are omitted by default; pass `log_bodies=True` to include them (still redacted, and this emits one WARNING at construction since bodies can contain other sensitive data).
+
 ### App configuration
 
 ```python

getstream/__init__.py

@@ -1,3 +1,5 @@
+import logging
+
 from getstream.exceptions import (  # noqa: F401
     StreamApiException,
     StreamException,
@@ -7,3 +9,7 @@
 )
 from getstream.stream import Stream  # noqa: F401
 from getstream.stream import AsyncStream  # noqa: F401
+
+# No-op until the caller attaches a handler: the SDK never configures the
+# logger's level or output, it only emits at each event's documented level.
+logging.getLogger("getstream").addHandler(logging.NullHandler())

getstream/base.py

@@ -13,10 +13,12 @@
     build_api_exception,
     wrap_transport_error,
 )
+from getstream.logging_utils import redact_json_body, redact_query
 from getstream.stream_response import StreamResponse
 from getstream.generic import T
 import httpx
 from getstream.config import BaseConfig
+from getstream.version import VERSION
 from urllib.parse import quote
 from abc import ABC
 from getstream.common.telemetry import (
@@ -59,22 +61,46 @@ def _resolve_pool_knobs(obj):
     )
 
 
-def _log_pool_config(cfg, *, user_http_client: bool) -> None:
-    if user_http_client:
-        logger.info(
-            "getstream connection pool: user_http_client=True (5 knobs not applied)"
-        )
-    else:
-        logger.info(
-            "getstream connection pool: "
-            "max_conns_per_host=%s idle_timeout=%ss "
-            "connect_timeout=%ss request_timeout=%ss "
-            "user_http_client=False",
-            cfg.max_conns_per_host,
-            cfg.idle_timeout,
-            cfg.connect_timeout,
-            cfg.timeout,
-        )
+def _resolve_logger(obj) -> logging.Logger:
+    """The caller's injected logger (``Stream``/``AsyncStream``'s ``logger=``
+    kwarg, plumbed onto ``obj.log`` the same way as the pool knobs), or the
+    shared module logger when none was passed."""
+    return getattr(obj, "log", None) or logger
+
+
+def _log_client_initialized(cfg, *, user_http_client: bool) -> None:
+    """Emit the one-shot ``client.initialized`` event, replacing the old
+    plain-text pool-config INFO line with the structured logging schema."""
+    _resolve_logger(cfg).info(
+        "client.initialized",
+        extra={
+            "stream.sdk.name": "stream-py",
+            "stream.sdk.version": VERSION,
+            "stream.client.max_conns_per_host": cfg.max_conns_per_host,
+            "stream.client.idle_timeout_seconds": cfg.idle_timeout,
+            "stream.client.connect_timeout_seconds": cfg.connect_timeout,
+            "stream.client.request_timeout_seconds": cfg.timeout,
+            "stream.client.gzip_enabled": not user_http_client,
+            "stream.client.user_http_client": user_http_client,
+            "stream.client.log_bodies": bool(getattr(cfg, "log_bodies", False)),
+        },
+    )
+
+
+def _response_body_for_log(response: httpx.Response):
+    """Redact a response body for the ``http.response.body`` log field.
+    JSON bodies get the shallow key redaction; anything else (or anything
+    that fails to parse as JSON) is passed through as text."""
+    # Media types are case-insensitive; match application/json and any
+    # structured +json type (e.g. application/problem+json) so their bodies
+    # are redacted, not logged as raw text.
+    content_type = response.headers.get("content-type", "").lower()
+    if "application/json" in content_type or "+json" in content_type:
+        try:
+            return redact_json_body(json.loads(response.text))
+        except (ValueError, TypeError):
+            return response.text
+    return response.text
 
 
 def _read_file_bytes(file_path: str) -> bytes:
@@ -284,6 +310,18 @@ def _request_sync(
         url_path, url_full, endpoint, attrs = self._prepare_request(
             method, path, query_params, kwargs
         )
+        log = _resolve_logger(self)
+        log_bodies = bool(getattr(self, "log_bodies", False))
+        sent_extra = {
+            "http.request.method": method,
+            "url.path": path,
+            "url.query": redact_query(query_params) or "",
+            "stream.endpoint_name": endpoint,
+        }
+        if log_bodies:
+            sent_extra["http.request.body"] = redact_json_body(kwargs.get("json"))
+        log.debug("http.request.sent", extra=sent_extra)
+
         start = time.perf_counter()
         # Span name uses logical operation (endpoint) rather than raw HTTP
         with span_request(
@@ -296,7 +334,19 @@ def _request_sync(
                     url_path, params=query_params, *args, **call_kwargs
                 )
             except httpx.RequestError as err:
-                raise wrap_transport_error(err) from err
+                exc = wrap_transport_error(err)
+                log.error(
+                    "http.request.failed",
+                    extra={
+                        "http.request.method": method,
+                        "url.path": path,
+                        "stream.endpoint_name": endpoint,
+                        "error.type": exc.error_type,
+                        "error.message": str(err),
+                        "duration_ms": int((time.perf_counter() - start) * 1000.0),
+                    },
+                )
+                raise exc from err
             duration = parse_duration_from_body(response.content)
             if duration:
                 span.set_attribute("http.server.duration", duration)
@@ -308,6 +358,17 @@ def _request_sync(
                 pass
 
             duration_ms = (time.perf_counter() - start) * 1000.0
+            received_extra = {
+                "http.request.method": method,
+                "url.path": path,
+                "stream.endpoint_name": endpoint,
+                "http.response.status_code": response.status_code,
+                "http.response.body.size": len(response.content or b""),
+                "duration_ms": int(duration_ms),
+            }
+            if log_bodies:
+                received_extra["http.response.body"] = _response_body_for_log(response)
+            log.debug("http.response.received", extra=received_extra)
             # Metrics should be low-cardinality: exclude url/call_cid/channel_cid
             metric_attrs = metric_attributes(
                 api_key=self.api_key,
@@ -593,6 +654,18 @@ async def _request_async(
         url_path, url_full, endpoint, attrs = self._prepare_request(
             method, path, query_params, kwargs
         )
+        log = _resolve_logger(self)
+        log_bodies = bool(getattr(self, "log_bodies", False))
+        sent_extra = {
+            "http.request.method": method,
+            "url.path": path,
+            "url.query": redact_query(query_params) or "",
+            "stream.endpoint_name": endpoint,
+        }
+        if log_bodies:
+            sent_extra["http.request.body"] = redact_json_body(kwargs.get("json"))
+        log.debug("http.request.sent", extra=sent_extra)
+
         start = time.perf_counter()
         with span_request(
             endpoint, attributes=attrs, request_body=kwargs.get("json")
@@ -612,7 +685,19 @@ async def _request_async(
                     url_path, params=query_params, *args, **call_kwargs
                 )
             except httpx.RequestError as err:
-                raise wrap_transport_error(err) from err
+                exc = wrap_transport_error(err)
+                log.error(
+                    "http.request.failed",
+                    extra={
+                        "http.request.method": method,
+                        "url.path": path,
+                        "stream.endpoint_name": endpoint,
+                        "error.type": exc.error_type,
+                        "error.message": str(err),
+                        "duration_ms": int((time.perf_counter() - start) * 1000.0),
+                    },
+                )
+                raise exc from err
             duration = parse_duration_from_body(response.content)
             if duration:
                 span.set_attribute("http.server.duration", duration)
@@ -624,6 +709,19 @@ async def _request_async(
                 pass
 
             duration_ms = (time.perf_counter() - start) * 1000.0
+            received_extra = {
+                "http.request.method": method,
+                "url.path": path,
+                "stream.endpoint_name": endpoint,
+                "http.response.status_code": response.status_code,
+                "http.response.body.size": len(response.content or b""),
+                "duration_ms": int(duration_ms),
+            }
+            if log_bodies:
+                received_extra["http.response.body"] = await asyncio.to_thread(
+                    _response_body_for_log, response
+                )
+            log.debug("http.response.received", extra=received_extra)
             # Metrics should be low-cardinality: exclude url/call_cid/channel_cid
             metric_attrs = metric_attributes(
                 api_key=self.api_key,

getstream/logging_utils.py

@@ -0,0 +1,21 @@
+"""Redaction helpers for the SDK's structured log events. Secret values are
+replaced with a literal marker; the scan is shallow by design."""
+
+REDACTED = "<redacted>"
+REDACTED_QUERY_PARAMS = {"api_key", "api_secret", "token"}
+REDACTED_BODY_KEYS = {"api_secret", "token", "password"}
+
+
+def redact_query(params):
+    if not params:
+        return params
+    return {
+        k: (REDACTED if k.lower() in REDACTED_QUERY_PARAMS else v)
+        for k, v in dict(params).items()
+    }
+
+
+def redact_json_body(body):
+    if not isinstance(body, dict):
+        return body
+    return {k: (REDACTED if k in REDACTED_BODY_KEYS else v) for k, v in body.items()}

getstream/stream.py

@@ -2,6 +2,7 @@
 
 from contextlib import AsyncExitStack
 from functools import cached_property
+import logging
 import time
 from typing import List, Optional
 from uuid import uuid4
@@ -10,7 +11,7 @@
 import jwt
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-from getstream.base import _log_pool_config
+from getstream.base import _log_client_initialized, _resolve_logger
 from getstream.common import telemetry
 from getstream.chat.client import ChatClient
 from getstream.chat.async_client import ChatClient as AsyncChatClient
@@ -92,6 +93,8 @@ def __init__(
         max_conns_per_host: Optional[int] = None,
         idle_timeout: Optional[float] = None,
         connect_timeout: Optional[float] = None,
+        logger: Optional[logging.Logger] = None,
+        log_bodies: bool = False,
     ):
         """Build a Stream client.
 
@@ -114,6 +117,8 @@ def __init__(
             max_conns_per_host: Max concurrent TCP connections per host. Default 5. Ignored when ``http_client`` is set.
             idle_timeout: Idle connection lifetime in seconds. Default 55.0 (sits 5s under the typical 60s LB idle timeout). Ignored when ``http_client`` is set.
             connect_timeout: TCP + TLS handshake timeout in seconds. Default 10.0. Ignored when ``http_client`` is set.
+            logger: Optional stdlib ``logging.Logger`` for the SDK's structured log events (``client.initialized``, ``http.request.sent``, ``http.response.received``, ``http.request.failed``). Defaults to ``logging.getLogger("getstream")``, which is a no-op until the caller attaches a handler.
+            log_bodies: When ``True``, adds redacted request/response bodies to the request/response log events. Off by default. Emits one WARNING at construction when enabled.
 
         Raises:
             ValueError: If both ``transport`` and ``http_client`` are set; if neither ``api_secret`` nor ``token`` can be resolved; if both are provided; if either is the empty string; if ``api_key`` is missing; or if ``request_timeout`` is not a positive number.
@@ -189,6 +194,13 @@ def _settings() -> _PoolSettings:
         self._transport = transport
         self._http_client = http_client
         self.token = token or self._create_token()
+        # log / log_bodies are read by BaseClient via getattr(self, ...), same
+        # plumbing as the pool knobs below: the intermediate generated REST
+        # clients (CommonRestClient etc.) do not forward these kwargs, so they
+        # must be set on self before super().__init__() and copied onto
+        # sub-clients in _apply_shared_client.
+        self.log = logger
+        self.log_bodies = log_bodies
         # Pool knobs are read by BaseClient via getattr(self, ...) since the intermediate generated REST clients (CommonRestClient etc.) do not forward these kwargs. self.max_conns_per_host / idle_timeout / connect_timeout were set above before super().__init__().
         super().__init__(
             self.api_key, self.base_url, self.token, self.timeout, self.user_agent
@@ -203,9 +215,17 @@ def _settings() -> _PoolSettings:
         # the parent's client avoids that and keeps one pool per Stream.
         self._shared_client = self.client
 
-        # Emit the pool-config INFO line exactly once per Stream, reflecting the
-        # resolved knobs on the top-level client. Sub-clients no longer log.
-        _log_pool_config(self, user_http_client=http_client is not None)
+        # Emit the client.initialized event exactly once per Stream, reflecting
+        # the resolved knobs on the top-level client. Sub-clients no longer log
+        # their own construction.
+        _log_client_initialized(self, user_http_client=http_client is not None)
+        if self.log_bodies:
+            _resolve_logger(self).warning(
+                "HTTP request/response bodies will be logged. Auth headers "
+                "and known-secret fields are still redacted, but other "
+                "sensitive data (messages, PII) may appear in logs. Disable "
+                "for production."
+            )
 
     @property
     def api_secret(self) -> str:
@@ -234,6 +254,11 @@ def _apply_shared_client(self, sub_client):
                 sub_client.client.close()
             sub_client.client = self._shared_client
             sub_client._owns_http_client = False
+        # log / log_bodies: same getattr(self, ..., default) plumbing as the
+        # pool knobs, so sub-clients (which issue the actual requests) emit
+        # through the caller's logger instead of silently falling back.
+        sub_client.log = getattr(self, "log", None)
+        sub_client.log_bodies = getattr(self, "log_bodies", False)
         return sub_client
 
     def create_token(
@@ -284,6 +309,8 @@ def clone_for_token(self, token: str):
             idle_timeout=self.idle_timeout,
             connect_timeout=self.connect_timeout,
             user_agent=self.user_agent,
+            logger=self.log,
+            log_bodies=self.log_bodies,
         )
 
     def create_call_token(
@@ -543,6 +570,8 @@ def as_async(self) -> "AsyncStream":
             connect_timeout=self.connect_timeout,
             base_url=self.base_url,
             user_agent=self.user_agent,
+            logger=self.log,
+            log_bodies=self.log_bodies,
         )
 
     @cached_property