Skip to content

feat: [CHA-2956] connection pooling: stream-py#260

Merged
mogita merged 23 commits into
mainfrom
feat/cha-2956_connection-pooling
May 28, 2026
Merged

feat: [CHA-2956] connection pooling: stream-py#260
mogita merged 23 commits into
mainfrom
feat/cha-2956_connection-pooling

Conversation

@mogita
Copy link
Copy Markdown
Contributor

@mogita mogita commented May 26, 2026

Summary

Implements explicit HTTP connection pooling for stream-py.

  • 4 new kwargs on Stream(...) and AsyncStream(...): max_conns_per_host, idle_timeout, connect_timeout, request_timeout
  • Defaults: 5 / 55.0s / 10.0s / 30.0s
  • INFO log on construction via the getstream logger
  • http_client= escape hatch unchanged — bypasses all 4 new kwargs
  • Sync/async parity locked by test
  • Per-call timeout= continues to work through .get/.post/... and pre-empts client request_timeout

Behavior change

Default request_timeout moves from 6.0s -> 30.0s. Existing callers using timeout= are unaffected — timeout is kept as an alias for request_timeout. CHANGELOG calls this out under "Changed."

Implementation note

The 3 knobs cannot pass through BaseStream.super().__init__() directly into BaseClient because that path goes through generated REST clients (CommonRestClient, etc.) which do not accept the new kwargs. To avoid touching generated code, BaseStream sets the knobs as attributes on self before super().__init__(), and BaseClient/AsyncBaseClient read them via a _resolve_pool_knobs(self) helper that falls back to defaults when the attributes are absent (covers directly-instantiated sub-clients and test fixtures).

Test plan

  • Default httpx.Limits (max_connections=5, max_keepalive_connections=5, keepalive_expiry=55.0)
  • Default httpx.Timeout (connect=10, read/write/pool=30)
  • Each knob individually overridable (sync + async)
  • Legacy timeout= alias still works
  • http_client= escape hatch (pool knobs not applied to user-supplied client)
  • INFO log with and without user-supplied http_client
  • Per-call timeout=httpx.Timeout(...) reaches httpx for both sync and async
  • Sync/async parity (same kwargs -> same pool + same timeout)
  • make test clean (565 passed, 41 skipped)
  • make lint clean
  • make typecheck — pre-existing failures in video/rtc/track_util.py only; no new errors introduced

Refs

Summary by CodeRabbit

  • New Features

    • Configurable HTTP connection pooling (max connections per host, idle-socket timeout, connect timeout).
    • Default per-request timeout increased from 6.0s to 30.0s; per-call timeout still overrides client defaults.
    • New client INFO log reports effective pool/timeout configuration.
    • Stream/AsyncStream webhook helpers: instance methods for signature verification and parsing that no longer require passing an api_secret.
  • Documentation

    • CHANGELOG updated with new defaults, precedence rules, and webhook helper notes.
  • Chore

    • Package version bumped to 3.5.0.
  • Tests

    • Added sync/async tests for pool/timeouts, logging, and override behaviors.

Review Change Stack

@coderabbitai
Copy link
Copy Markdown

coderabbitai Bot commented May 26, 2026

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

  • @coderabbitai resume to resume automatic reviews.
  • @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

  • ▶️ Resume reviews
  • 🔍 Trigger review
📝 Walkthrough

Walkthrough

Adds request_timeout and connection-pool knobs to Stream, propagates them through Settings/BaseConfig, constructs internal httpx clients with httpx.Limits and structured httpx.Timeout when no user http_client is provided, logs effective pool config, updates CHANGELOG/version, and expands tests.

Changes

HTTP Connection Pooling Implementation

Layer / File(s) Summary
Module defaults and configuration schema
getstream/stream.py, getstream/config.py, getstream/base.py
Introduce DEFAULT_REQUEST_TIMEOUT, DEFAULT_MAX_CONNS_PER_HOST, DEFAULT_IDLE_TIMEOUT, DEFAULT_CONNECT_TIMEOUT; extend Settings with request_timeout and pool knobs; add BaseConfig params; add module logger and helpers _resolve_pool_knobs and _log_pool_config.
BaseStream initialization and timeout resolution
getstream/stream.py
BaseStream.__init__ gains request_timeout and pool-knob kwargs; timeout is a backward-compat alias; implements precedence (kwarg > alias > env-derived Settings > default), validates positive request_timeout, stores self.timeout/self.request_timeout, and propagates knobs in clone_for_token and as_async.
BaseClient and AsyncBaseClient HTTP pooling construction
getstream/base.py
Clients resolve pool knobs and forward them into BaseConfig; when SDK constructs an internal httpx.Client/httpx.AsyncClient it now uses httpx.Limits and a structured httpx.Timeout (connect/read/write/pool) and calls _log_pool_config after creation. External user-provided http_client path is unchanged.
Tests, docs, and version bump
tests/*, CHANGELOG.md, pyproject.toml
Add sync/async tests for default pool/timeouts, overrides, legacy timeout alias, escape-hatch (user http_client) behavior, INFO construction logs, parity and per-call timeout propagation, and _resolve_pool_knobs regressions. Update CHANGELOG to document knobs, defaults, env fallbacks, logging, and bump project version to 3.5.0; adjust video example tests and small webhook test comments.

🎯 4 (Complex) | ⏱️ ~45 minutes

"🐰
Pool knobs set, the connections sip and hum,
Five per host, idle sleeps fifty-five some,
Thirty seconds wait now where six once stood,
INFO logs whisper what the client built for good,
Hops and timeouts tuned — the rabbits nod and drum."

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name Status Explanation Resolution
Docstring Coverage ⚠️ Warning Docstring coverage is 11.11% which is insufficient. The required threshold is 80.00%. Write docstrings for the functions missing them to satisfy the coverage threshold.
✅ Passed checks (4 passed)
Check name Status Explanation
Description Check ✅ Passed Check skipped - CodeRabbit’s high-level summary is enabled.
Title check ✅ Passed The title clearly summarizes the main change: implementing connection pooling configuration for stream-py with specific reference to the feature type and ticket identifier.
Linked Issues check ✅ Passed Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check ✅ Passed Check skipped because no linked issues were found for this pull request.

✏️ Tip: You can configure your own custom pre-merge checks in the settings.

✨ Finishing Touches
🧪 Generate unit tests (beta)
  • Create PR with unit tests
  • Commit unit tests in branch feat/cha-2956_connection-pooling

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

Comment @coderabbitai help to get the list of available commands and usage tips.

Copy link
Copy Markdown

@coderabbitai coderabbitai Bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (2)
getstream/stream.py (1)

274-280: ⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Forward pool configuration when cloning/converting clients.

These constructors carry timeout but drop max_conns_per_host, idle_timeout, and connect_timeout, so cloned/async-converted clients can silently revert to defaults instead of preserving the source client config.

Suggested fix
     def clone_for_token(self, token: str):
         ...
         return self.__class__(
             api_key=self.api_key,
             token=token,
             base_url=self.base_url,
-            timeout=self.timeout,
+            request_timeout=self.request_timeout,
+            max_conns_per_host=self.max_conns_per_host,
+            idle_timeout=self.idle_timeout,
+            connect_timeout=self.connect_timeout,
             user_agent=self.user_agent,
         )

     def as_async(self) -> "AsyncStream":
         return AsyncStream(
             api_key=self.api_key,
             api_secret=self._api_secret,
             token=None if self.has_api_secret else self.token,
-            timeout=self.timeout,
+            request_timeout=self.request_timeout,
+            max_conns_per_host=self.max_conns_per_host,
+            idle_timeout=self.idle_timeout,
+            connect_timeout=self.connect_timeout,
             base_url=self.base_url,
             user_agent=self.user_agent,
         )

Also applies to: 506-514

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@getstream/stream.py` around lines 274 - 280, The client clone/conversion is
not forwarding connection-pool settings, causing resets to defaults; update the
constructor call in the method that returns self.__class__(...) to include and
forward self.max_conns_per_host, self.idle_timeout, and self.connect_timeout
(and any other pool-related attributes present on the instance) so the
cloned/async-converted client preserves the original pool configuration; apply
the same change to the other similar constructor call elsewhere in the file that
performs client cloning/conversion.
getstream/base.py (1)

223-224: ⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Avoid overwriting caller-supplied http_client timeout.

When http_client is provided, getstream/base.py still mutates user-managed timeout via http_client.timeout = httpx.Timeout(self.timeout) in both sync (223-224) and async (486-487) paths.

Suggested fix
         if http_client is not None:
@@
-            if self.timeout is not None:
-                http_client.timeout = httpx.Timeout(self.timeout)
             self.client = http_client
             self._owns_http_client = False
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@getstream/base.py` around lines 223 - 224, The code currently mutates a
caller-supplied http_client by doing http_client.timeout =
httpx.Timeout(self.timeout); instead, detect if http_client was passed in and do
not modify it—only apply self.timeout when creating an internal client (e.g.
construct httpx.Client or httpx.AsyncClient with timeout=self.timeout) or by
using a copy/new client; in other words, remove the assignment to
http_client.timeout when an external http_client is provided and ensure internal
client creation paths use httpx.Timeout(self.timeout) so caller-managed
http_client remains unchanged.
🧹 Nitpick comments (1)
tests/test_http_client.py (1)

31-127: ⚡ Quick win

Use pytest fixtures for client creation in these new tests.

These blocks repeatedly instantiate Stream/AsyncStream inline; please move creation/teardown into shared fixtures and inject them into tests to match repo test conventions.

As per coding guidelines, "Use fixtures to inject objects in tests; test client fixtures can use the Stream API client".

Also applies to: 292-343, 349-466

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_http_client.py` around lines 31 - 127, Several tests repeatedly
construct Stream and AsyncStream inline (see TestSyncPoolDefaults,
TestAsyncPoolDefaults, TestSyncPoolOverrides._make and
TestAsyncPoolOverrides._make and the individual test methods); refactor them to
use pytest fixtures that create and yield a Stream (e.g., fixture name
sync_client) and an AsyncStream (e.g., async_client) so creation and teardown
(client.aclose()) are centralized. Replace inline constructions in test methods
with injected fixtures (accept sync_client or async_client as parameters),
remove manual aclose calls in tests that will be handled in the fixture
teardown, and keep existing assertions unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@getstream/base.py`:
- Around line 50-53: The current _resolve_pool_knobs implementation uses
expressions like getattr(obj, "max_conns_per_host", None) or
DEFAULT_MAX_CONNS_PER_HOST which treats explicit falsy values (0/0.0) as
missing; update _resolve_pool_knobs to only substitute the DEFAULT_* values when
getattr(...) returns None (e.g., use a conditional that checks "is None" or the
ternary form) for max_conns_per_host, idle_timeout, and connect_timeout so
caller-provided 0/0.0 are preserved.

---

Outside diff comments:
In `@getstream/base.py`:
- Around line 223-224: The code currently mutates a caller-supplied http_client
by doing http_client.timeout = httpx.Timeout(self.timeout); instead, detect if
http_client was passed in and do not modify it—only apply self.timeout when
creating an internal client (e.g. construct httpx.Client or httpx.AsyncClient
with timeout=self.timeout) or by using a copy/new client; in other words, remove
the assignment to http_client.timeout when an external http_client is provided
and ensure internal client creation paths use httpx.Timeout(self.timeout) so
caller-managed http_client remains unchanged.

In `@getstream/stream.py`:
- Around line 274-280: The client clone/conversion is not forwarding
connection-pool settings, causing resets to defaults; update the constructor
call in the method that returns self.__class__(...) to include and forward
self.max_conns_per_host, self.idle_timeout, and self.connect_timeout (and any
other pool-related attributes present on the instance) so the
cloned/async-converted client preserves the original pool configuration; apply
the same change to the other similar constructor call elsewhere in the file that
performs client cloning/conversion.

---

Nitpick comments:
In `@tests/test_http_client.py`:
- Around line 31-127: Several tests repeatedly construct Stream and AsyncStream
inline (see TestSyncPoolDefaults, TestAsyncPoolDefaults,
TestSyncPoolOverrides._make and TestAsyncPoolOverrides._make and the individual
test methods); refactor them to use pytest fixtures that create and yield a
Stream (e.g., fixture name sync_client) and an AsyncStream (e.g., async_client)
so creation and teardown (client.aclose()) are centralized. Replace inline
constructions in test methods with injected fixtures (accept sync_client or
async_client as parameters), remove manual aclose calls in tests that will be
handled in the fixture teardown, and keep existing assertions unchanged.
🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 0242630f-1bd3-42e0-9c1c-ff484b1d1ba0

📥 Commits

Reviewing files that changed from the base of the PR and between 91fdc5a and b87c9fb.

⛔ Files ignored due to path filters (1)
  • uv.lock is excluded by !**/*.lock
📒 Files selected for processing (6)
  • CHANGELOG.md
  • getstream/base.py
  • getstream/config.py
  • getstream/stream.py
  • pyproject.toml
  • tests/test_http_client.py

Comment thread getstream/base.py Outdated
Caller-provided falsy values (max_conns_per_host=0, idle_timeout=0.0)
were silently replaced with spec defaults because of the 'getattr(...)
or DEFAULT_*' pattern. Switch to 'is None' so the resolver only
substitutes a default when the attribute is missing or explicitly None.

Adds TestResolvePoolKnobsExplicitZero with three cases: explicit 0
preserved, missing attrs fall back, explicit None falls back.
test_setup_client and test_async_client asserted the old 6.0s default
timeout; CHA-2956 moved the default request timeout to 30s. The
http-client suite was updated in the original commit but these two
construction smoke tests were missed.
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)
getstream/base.py (1)

64-68: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Log message says "5 knobs" but there are only 4.

The PR introduces four new kwargs (max_conns_per_host, idle_timeout, connect_timeout, request_timeout), but the log message claims 5 are skipped.

Proposed fix
     if user_http_client:
         logger.info(
-            "getstream connection pool: user_http_client=True (5 knobs not applied)"
+            "getstream connection pool: user_http_client=True (4 knobs not applied)"
         )
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@getstream/base.py` around lines 64 - 68, The log in _log_pool_config
currently claims "5 knobs not applied" but only four kwargs were added
(max_conns_per_host, idle_timeout, connect_timeout, request_timeout); update the
message in _log_pool_config to reflect the correct count (e.g., "4 knobs not
applied") or list the four knobs explicitly so the log is accurate, ensuring the
change is made inside the _log_pool_config function where user_http_client is
checked.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Outside diff comments:
In `@getstream/base.py`:
- Around line 64-68: The log in _log_pool_config currently claims "5 knobs not
applied" but only four kwargs were added (max_conns_per_host, idle_timeout,
connect_timeout, request_timeout); update the message in _log_pool_config to
reflect the correct count (e.g., "4 knobs not applied") or list the four knobs
explicitly so the log is accurate, ensuring the change is made inside the
_log_pool_config function where user_http_client is checked.

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: a42eb14e-7c15-405e-a2ef-3dfd5c764a91

📥 Commits

Reviewing files that changed from the base of the PR and between 3c5f546 and 697b64c.

📒 Files selected for processing (6)
  • CHANGELOG.md
  • getstream/base.py
  • getstream/stream.py
  • getstream/tests/test_webhook.py
  • tests/test_http_client.py
  • tests/test_video_examples.py
✅ Files skipped from review due to trivial changes (2)
  • getstream/tests/test_webhook.py
  • CHANGELOG.md
🚧 Files skipped from review as they are similar to previous changes (2)
  • getstream/stream.py
  • tests/test_http_client.py

mogita added 5 commits May 27, 2026 12:18
Both built fresh top-level clients passing only timeout, so cloned and
async-converted clients silently reverted to default pool config. Forward
request_timeout + max_conns_per_host + idle_timeout + connect_timeout.
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick comments (1)
tests/test_http_client.py (1)

529-538: ⚡ Quick win

Use a pytest fixture instead of a helper constructor in tests.

Lines 529-538 currently use _make() for object setup. Please switch this to a fixture so setup is injected consistently across tests.

♻️ Suggested refactor
 class TestPoolConfigForwardedOnClone:
@@
-    def _make(self):
-        return Stream(
+    `@pytest.fixture`
+    def stream_client(self):
+        client = Stream(
             api_key="k",
             api_secret="s",
             base_url="http://test",
             max_conns_per_host=20,
             idle_timeout=120.0,
             connect_timeout=3.0,
             request_timeout=15.0,
         )
+        yield client
+        client.close()
 
-    def test_clone_for_token_preserves_pool_config(self):
-        clone = self._make().clone_for_token("user-token")
+    def test_clone_for_token_preserves_pool_config(self, stream_client):
+        clone = stream_client.clone_for_token("user-token")
@@
     `@pytest.mark.asyncio`
-    async def test_as_async_preserves_pool_config(self):
-        sync_client = self._make()
-        aclient = sync_client.as_async()
+    async def test_as_async_preserves_pool_config(self, stream_client):
+        aclient = stream_client.as_async()
@@
         assert aclient.request_timeout == 15.0
         assert aclient.client._transport._pool._max_connections == 20
         await aclient.aclose()
-        sync_client.close()

As per coding guidelines, "**/test_*.py: Use fixtures to inject objects in tests; test client fixtures can use the Stream API client".

Also applies to: 540-553

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_http_client.py` around lines 529 - 538, Replace the local helper
constructor _make() with a pytest fixture that yields a configured Stream client
instance so tests receive setup via injection: define a fixture (e.g.,
stream_client or stream) that returns Stream(api_key="k", api_secret="s",
base_url="http://test", max_conns_per_host=20, idle_timeout=120.0,
connect_timeout=3.0, request_timeout=15.0) and update tests that called _make()
to accept the fixture parameter instead; remove the _make() function and any
direct calls to it to ensure consistent fixture-based setup across tests (also
change occurrences around lines 540-553).
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@tests/test_http_client.py`:
- Around line 529-538: Replace the local helper constructor _make() with a
pytest fixture that yields a configured Stream client instance so tests receive
setup via injection: define a fixture (e.g., stream_client or stream) that
returns Stream(api_key="k", api_secret="s", base_url="http://test",
max_conns_per_host=20, idle_timeout=120.0, connect_timeout=3.0,
request_timeout=15.0) and update tests that called _make() to accept the fixture
parameter instead; remove the _make() function and any direct calls to it to
ensure consistent fixture-based setup across tests (also change occurrences
around lines 540-553).

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: c302ce5e-35c9-4ae7-ad01-36045a0a1a8b

📥 Commits

Reviewing files that changed from the base of the PR and between cfe2b02 and 0759b65.

📒 Files selected for processing (3)
  • CHANGELOG.md
  • getstream/stream.py
  • tests/test_http_client.py
✅ Files skipped from review due to trivial changes (1)
  • CHANGELOG.md

mogita added 6 commits May 27, 2026 15:31
Sub-clients (video/chat/moderation/feeds) are what issue the real API
calls, but on the default path they each built their own httpx client
via the generated REST clients, which do not forward the pool kwargs.
The pool knobs were therefore silently dropped and sub-client pools fell
back to the SDK defaults (5/55), making max_conns_per_host/idle_timeout/
connect_timeout a no-op for actual traffic.

Always share the parent's single fully-configured client with the
sub-clients (the same mechanism already used for the transport/http_client
paths), so the resolved pool config reaches the clients that send requests.
The pool-knob / request-timeout env fallbacks instantiated the full
Settings model whenever a knob or timeout was unset (the normal case).
Settings.api_key is a required field, so Settings() raised a pydantic
ValidationError in any environment without STREAM_* vars, crashing
Stream(api_key="k", api_secret="s") in a clean environment. The test
suite hid this because the fixtures call load_dotenv().

Introduce a small _PoolSettings model that reads the same STREAM_* env
vars but has no required api_key, and use it for knob resolution. Existing
env behavior (including the STREAM_TIMEOUT alias) is unchanged.
getstream/tests/test_webhook.py carries a "Code generated ... DO NOT
EDIT." header but was modified by the em-dash cleanup. Restore it to
origin/main so it leaves this PR's diff; the generated content is handled
via the chat/ template separately.
The pool-config INFO line was emitted from every BaseClient/AsyncBaseClient
construction, so a single Stream logged it once per sub-client (about five
lines) and the sub-client lines reported the default knobs rather than the
configured ones. Emit it exactly once from BaseStream after the top-level
client is built, reflecting the resolved knobs, and drop the per-sub-client
emission.

Also default Stream.from_env(timeout=...) to None so it inherits the new
30.0s default request timeout instead of the old hard-coded 6.0s.
Applied the em-dash delta directly because a clean regen is blocked by a
pre-existing generator drift (duplicate AsyncExportErrorEvent imports);
the template fix is in chat/ cha-2956_sdk-templates, so this matches what
a future clean regen will produce.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants