Provide Server-Sent Events handler.

class http_sse.Event(data: str = '', type: str = 'message', last_event_id: str = '', origin: str = '')

Bases: object

Event represents a message received.

data: str = ''

The data sent by the message emitter.

last_event_id: str = ''

A string representing a unique ID for the event.

origin: str = ''

A string representing the origin of the message emitter.

type: str = 'message'

The name identifying the type of the event.

class http_sse.EventStream(url: str, *, method: str = 'GET', headers: dict | None = None, timeout: float = 30, request_kwargs: dict | None = None, session: Session | None = None, reconnection_time: float = 30, max_connect_retry: float = 3, last_event_id: str = '')

Bases: object

EventStream wraps a persistent connection to an HTTP server, which sends events in text/event-stream format.

Usage:

from http_sse import EventStream

with EventStream("https://sse.dev/") as ev_stream:
    for event in ev_stream:
        print(event)

start() → None

Open the HTTP connection.

This method retries to connect up to max_connect_retry times.

stop() → None

Close the HTTP connection.

After closing, this instance is no longer usable.