feat: Add AsyncSSEClient with aiohttp-based async/await support#58
feat: Add AsyncSSEClient with aiohttp-based async/await support#58
Conversation
Adds AsyncSSEClient as a purely additive new public API alongside the existing SSEClient. Async users install with the [async] extra to get aiohttp; sync users have no new dependencies. All existing tests pass unchanged. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 2 potential issues.
Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Removes make_stream, retry_for_status, and no_delay from async_helpers.py as they were either unused or duplicates of the same functions in helpers.py. Updates async test files to import these from helpers.py consistently. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
|
|
||
| request_options = {} | ||
| if self.options.get("readTimeoutMs") is not None: | ||
| request_options["timeout"] = aiohttp.ClientTimeout( |
There was a problem hiding this comment.
I think, even without a read timeout, we need to customize the ClientTimeout.
The timeout uses a non-default ClientTimeout.
But if we do set it to anything, then the total will get set to None. https://docs.aiohttp.org/en/stable/client_reference.html#aiohttp.ClientTimeout.total
But if we set a ClientTimeout, we would get the default total=None.
| error = e | ||
| self.__connection_result = None | ||
| finally: | ||
| self.__last_event_id = reader.last_event_id |
There was a problem hiding this comment.
Should we close the result in finally out of an abundance of caution? By the docs you normally don't need to close it if it completes normally, and it would be closed on a network error, but the exception here here could also potentially be a decoding error, in which maybe we wouldn't close it?
|
|
||
| async def close(self): | ||
| # Only close the session if we created it ourselves | ||
| if self.__external_session is None and self.__session is not None: |
There was a problem hiding this comment.
I think the non-async version does this wrong?
python-eventsource/ld_eventsource/http.py
Line 60 in d086a6a
self.__should_close_pool = params.pool is not None
Not a problem with this PR, but maybe worth checking.
2 participants
Adds AsyncSSEClient as a purely additive new public API alongside the
existing SSEClient. Async users install with the [async] extra to get
aiohttp; sync users have no new dependencies. All existing tests pass
unchanged.
Note
Medium Risk
Adds a new async client and HTTP stack (
aiohttp) with new retry/streaming behavior and contract-test infrastructure; although largely additive and behind an optional extra, it introduces new concurrency/resource-lifecycle paths that could affect reliability if misused.Overview
Adds a new public async/await SSE API (
AsyncSSEClient) built aroundaiohttp, including async stream parsing (_AsyncBufferedLineReader/_AsyncSSEReader), an async connection abstraction (AsyncConnectStrategy), and an async HTTP connector with header/query-param support and response validation.Updates packaging and docs to keep async dependencies optional (
launchdarkly-eventsource[async]) and lazily exposesAsyncSSEClientviald_eventsource.__getattr__to avoid importingaiohttpfor sync-only users.Extends test coverage and automation by adding pytest-asyncio-based unit tests, an async contract-test service (
contract-tests/async_service.py), Makefile targets, and CI steps to run the async contract tests; also mocksaiohttpduring doc builds.Written by Cursor Bugbot for commit 069bffe. This will update automatically on new commits. Configure here.