sseclient-py is a Python library providing a client for Server-Sent Events (SSE). It simplifies consuming event streams by parsing the SSE protocol over HTTP. The current stable version is 1.9.0, with releases typically focused on bug fixes and feature additions rather than frequent breaking changes.
pip install sseclient-pyVerified import paths — ran on the pinned version, not inferred.
Connects to an SSE stream using the `sseclient-py` library with `requests`. It demonstrates how to initialize the client and iterate over incoming events, highlighting the crucial `stream=True` parameter for `requests` to ensure proper streaming behavior.
Ensure your `requests.get()` call includes the `stream=True` parameter, e.g., `requests.get(url, stream=True)`.
Wrap the SSE consumption loop in a retry mechanism (e.g., a `while True` loop with `try...except` and a backoff delay) to re-establish the connection upon failure.
Ensure the SSE server provides well-formed events. Implement robust error handling around event processing, and consider logging raw data for debugging malformed streams.
Install the library using pip: ```bash pip install sseclient-py ```
Use the `requests` library to make an HTTP GET request to the SSE endpoint and pass the resulting `requests.Response` object to `sseclient.SSEClient`:
```python
import requests
import sseclient
response = requests.get('http://example.com/sse', stream=True)
client = sseclient.SSEClient(response)
```Always include `stream=True` in your `requests.get()` call when consuming an SSE stream:
```python
import requests
import sseclient
response = requests.get('http://example.com/sse', stream=True)
client = sseclient.SSEClient(response)
for event in client.events():
print(event.data)
```Access the event's payload using `event.data`. If the data is JSON, parse it explicitly using `json.loads()`:
```python
import json
import requests
import sseclient
response = requests.get('http://example.com/sse', stream=True)
client = sseclient.SSEClient(response)
for event in client.events():
if event.data:
try:
parsed_data = json.loads(event.data)
print('Parsed JSON data:', parsed_data)
except json.JSONDecodeError:
print('Raw data:', event.data)
```No dependency data recorded yet.