feat: [CHA-2956] connection pooling: stream-py#260
Open
mogita wants to merge 23 commits into
Open
Conversation
|
Note Reviews pausedIt 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 Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds 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. ChangesHTTP Connection Pooling Implementation
🎯 4 (Complex) | ⏱️ ~45 minutes
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
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 winForward pool configuration when cloning/converting clients.
These constructors carry
timeoutbut dropmax_conns_per_host,idle_timeout, andconnect_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 winAvoid overwriting caller-supplied http_client timeout.
When
http_clientis provided,getstream/base.pystill mutates user-managed timeout viahttp_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 winUse pytest fixtures for client creation in these new tests.
These blocks repeatedly instantiate
Stream/AsyncStreaminline; 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
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (6)
CHANGELOG.mdgetstream/base.pygetstream/config.pygetstream/stream.pypyproject.tomltests/test_http_client.py
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.
There was a problem hiding this comment.
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 winLog 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
📒 Files selected for processing (6)
CHANGELOG.mdgetstream/base.pygetstream/stream.pygetstream/tests/test_webhook.pytests/test_http_client.pytests/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
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.
There was a problem hiding this comment.
🧹 Nitpick comments (1)
tests/test_http_client.py (1)
529-538: ⚡ Quick winUse 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
📒 Files selected for processing (3)
CHANGELOG.mdgetstream/stream.pytests/test_http_client.py
✅ Files skipped from review due to trivial changes (1)
- CHANGELOG.md
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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements explicit HTTP connection pooling for stream-py.
Stream(...)andAsyncStream(...):max_conns_per_host,idle_timeout,connect_timeout,request_timeoutgetstreamloggerhttp_client=escape hatch unchanged — bypasses all 4 new kwargstimeout=continues to work through.get/.post/...and pre-empts clientrequest_timeoutBehavior change
Default
request_timeoutmoves from 6.0s -> 30.0s. Existing callers usingtimeout=are unaffected —timeoutis kept as an alias forrequest_timeout. CHANGELOG calls this out under "Changed."Implementation note
The 3 knobs cannot pass through
BaseStream.super().__init__()directly intoBaseClientbecause that path goes through generated REST clients (CommonRestClient, etc.) which do not accept the new kwargs. To avoid touching generated code,BaseStreamsets the knobs as attributes onselfbeforesuper().__init__(), andBaseClient/AsyncBaseClientread 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
httpx.Limits(max_connections=5, max_keepalive_connections=5, keepalive_expiry=55.0)httpx.Timeout(connect=10, read/write/pool=30)timeout=alias still workshttp_client=escape hatch (pool knobs not applied to user-supplied client)http_clienttimeout=httpx.Timeout(...)reaches httpx for both sync and asyncmake testclean (565 passed, 41 skipped)make lintcleanmake typecheck— pre-existing failures invideo/rtc/track_util.pyonly; no new errors introducedRefs
Summary by CodeRabbit
New Features
Documentation
Chore
Tests