Server-Sent Events (SSE)

class microdot.sse.SSE

Server-Sent Events object.

An object of this class is sent to handler functions to manage the SSE connection.

async send(data, event=None, event_id=None, retry=None, comment=False)

Send an event to the client.

Parameters:

  • data – the data to send. It can be given as a string, bytes, dict or list. Dictionaries and lists are serialized to JSON. Any other types are converted to string before sending.

  • event – an optional event name, to send along with the data. If given, it must be a string.

  • event_id – an optional event id, to send along with the data. If given, it must be a string.

  • retry – an optional reconnection time (in seconds) that the client should use when the connection is lost.

  • comment – when set to True, the data is sent as a comment line, and all other parameters are ignored. This is useful as a heartbeat mechanism that keeps the connection alive.

microdot.sse.sse_response(request, event_function, *args, **kwargs)

Return a response object that initiates an event stream.

Parameters:

  • request – the request object.

  • event_function – an asynchronous function that will send events to the client. The function is invoked with request and an sse object. The function should use sse.send() to send events to the client.

  • args – additional positional arguments to be passed to the response.

  • kwargs – additional keyword arguments to be passed to the response.

This is a low-level function that can be used to implement a custom SSE endpoint. In general the microdot.sse.with_sse() decorator should be used instead.

microdot.sse.with_sse(f)

Decorator to make a route a Server-Sent Events endpoint.

This decorator is used to define a route that accepts SSE connections. The route then receives a sse object as a second argument that it can use to send events to the client:

@app.route('/events')
@with_sse
async def events(request, sse):
    # send an unnamed event with string data
    await sse.send('hello')

    # send an unnamed event with JSON data
    await sse.send({'foo': 'bar'})

    # send a named event
    await sse.send('hello', event='greeting')