LaunchDarkly SSE Client

1.5.1 · active · verified Wed Apr 08

The `launchdarkly-eventsource` library is a Python client for Server-Sent Events (SSE), enabling applications to subscribe to and process real-time events from an SSE server. It is currently at version 1.5.1 and maintains an active release cadence with regular updates and bug fixes, typically every few months.

Warnings

breaking Python 3.8 is no longer supported starting with version 1.3.0. Attempts to install or run on Python 3.8 will fail or result in unexpected behavior.
Fix: Upgrade your Python environment to 3.9 or higher. Python 3.8 reached end-of-life on October 7, 2024.
breaking Python 3.7 is no longer supported starting with version 1.2.0. Users on Python 3.7 must upgrade their Python version to use this library.
Fix: Upgrade your Python environment to 3.8 or higher (and preferably 3.9+ for compatibility with newer versions).
gotcha Older versions (prior to 1.2.3) might encounter issues with deprecated `ssl.wrap_socket` usage when dealing with custom SSL contexts or certain environments, potentially leading to connection failures.
Fix: Ensure you are using `launchdarkly-eventsource` version 1.2.3 or newer to leverage the fix for `ssl.wrap_socket` usage.

Install

pip install launchdarkly-eventsource Install stable version

Imports

EventSource
```
from ld_eventsource import EventSource
```
MessageEvent
```
from ld_eventsource import MessageEvent
```
Used for type hinting and accessing event details.

Quickstart

This quickstart demonstrates how to connect to an SSE endpoint, register handlers for message, error, open, and close events, and then start listening for events. The `start()` method is blocking, so in a long-running application, it's typically run in a separate thread. For this example, it will attempt to connect and process events until closed.

import os
import time
from ld_eventsource import EventSource, MessageEvent

# Replace with your SSE endpoint
SSE_URL = os.environ.get('SSE_ENDPOINT', 'http://localhost:8080/events')

def handle_event(event: MessageEvent):
    print(f"Received event: ID={event.last_event_id}, Type={event.event_type}, Data={event.data}")

def handle_error(error):
    print(f"Error: {error}")

def handle_open():
    print("Connection opened.")

def handle_closed():
    print("Connection closed.")

def main():
    print(f"Connecting to SSE_URL: {SSE_URL}")
    es = None
    try:
        es = EventSource(SSE_URL)
        es.on_message += handle_event
        es.on_error += handle_error
        es.on_open += handle_open
        es.on_closed += handle_closed

        print("Starting EventSource (listening for 10 seconds for demonstration)...")
        # EventSource.start() blocks until connection closes or .close() is called.
        # In a real application, you might run this in a separate thread.
        es.start()

    except Exception as e:
        print(f"An unexpected error occurred: {e}")
    finally:
        if es and es.is_running:
            print("Closing EventSource connection.")
            es.close()
        print("Exiting.")

if __name__ == "__main__":
    main()

view raw JSON →