Assel

A sans I/O Server-Sent Events (SSE) library. This library focuses on parsing and creating SSE event streams but does implement an a complete EventSource. This means for example that the reconnection time, if present, is parsed and accessible to you, but the library does not help you with acting on it, ie. reconnecting.

Features:

• parse and serialize (receive and send) SSE events
• send comments
• stream API
• usable in sync and async contexts

Usage

[!NOTE] Documentation is WIP. See tests for more usage examples

There is a lower level "sans-io" API dealing with bytes and lists of Events and a higher level (a)sync stream API dealing with streams of bytes and Events

Parser API

# chunked up bytes stream as you (may) get it from IO
chunks = [b"da", b"ta: hel", b"lo\n\n", b"event: add\ndata: 5\n\ndata: hello\n\n"]

p = sse.Parser()
for chunk in chunks:
    events = p.receive_bytes(chunk)
    for event in events:
        print(event)

should print:

Event(data='hello', event='message')
Event(data='5', event='add')
Event(data='hello', event='message')

stream API

import assel as sse

# chunked up bytes stream as you get it from IO
chunks = [b"da", b"ta: hel", b"lo\n\n", b"event: add\ndata: 5\n\n"]
for event in sse.parse(chunks):
    print(event)

sending events

To send events, pass an Event to the send function. This will produce bytes suitable for giving to IO operations as part of an SSE stream. You may attach comments to events or generate "stand-alone" comment lines.

# prints: b'event: add\ndata: 5\n\n'
print(sse.send(sse.Event("5", event="add")))

# prints: b':foo\ndata: hello\n\n'
print(sse.send(event=sse.Event("hello"), comment="foo"))

# prints: b':foo\n'
print(sse.send(comment="foo"))

Development

uv run pytest

Acknowledgements

This software was originally created as part of my work for the "Computing" Working Group of the Gesellschaft für wissenschaftliche Datenverarbeitung mbH Göttingen (GWDG) during my employment at the University of Göttingen.