Skip to content

API Reference

sse_stream

django_flosse.decorators.sse_stream(_view_func: Optional[Callable] = None, *, retry: Optional[int] = None) -> Callable

Decorator that turns a generator view into an SSE endpoint.

Supports both sync and async generator views — detected automatically::

# sync
@sse_stream
def feed(request):
    yield ("update", {"value": 1})

# async — requires ASGI server (Uvicorn, Daphne)
@sse_stream
async def feed(request):
    async for item in my_async_source():
        yield ("update", {"value": item})

Can be used with or without parentheses::

@sse_stream
@sse_stream()
@sse_stream(retry=3000)

Parameters:

Name Type Description Default
retry Optional[int]

If given, sends a retry: directive (in ms) as the first frame, telling the browser how long to wait before reconnecting.

 None
Yield styles
  • str → plain data event
  • (event_name, data) → named event
  • (event_name, data, id) → named event with ID
  • dict with "data" key → mapped to SSEEvent fields
  • dict without "data" → whole dict serialised as JSON
  • :class:~django_flosse.events.SSEEvent → used directly
Heartbeats

django-flosse does not manage heartbeats automatically. Emit a keep-alive ping yourself if behind a proxy::

@sse_stream
async def live_feed(request):
    while True:
        data = await get_new_data()
        if data:
            yield ("update", data)
        else:
            yield SSEEvent(data="", event="ping")
        await asyncio.sleep(5)
Source code in django_flosse/decorators.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def sse_stream(
    _view_func: Optional[Callable] = None,
    *,
    retry: Optional[int] = None,
) -> Callable:
    """
    Decorator that turns a **generator view** into an SSE endpoint.

    Supports both sync and async generator views — detected automatically::

        # sync
        @sse_stream
        def feed(request):
            yield ("update", {"value": 1})

        # async — requires ASGI server (Uvicorn, Daphne)
        @sse_stream
        async def feed(request):
            async for item in my_async_source():
                yield ("update", {"value": item})

    Can be used with or without parentheses::

        @sse_stream
        @sse_stream()
        @sse_stream(retry=3000)

    Parameters
    ----------
    retry:
        If given, sends a ``retry:`` directive (in ms) as the first frame,
        telling the browser how long to wait before reconnecting.

    Yield styles
    ------------
    * ``str``                        → plain data event
    * ``(event_name, data)``         → named event
    * ``(event_name, data, id)``     → named event with ID
    * ``dict`` with ``"data"`` key   → mapped to SSEEvent fields
    * ``dict`` without ``"data"``    → whole dict serialised as JSON
    * :class:`~django_flosse.events.SSEEvent` → used directly

    Heartbeats
    ----------
    django-flosse does **not** manage heartbeats automatically.
    Emit a keep-alive ping yourself if behind a proxy::

        @sse_stream
        async def live_feed(request):
            while True:
                data = await get_new_data()
                if data:
                    yield ("update", data)
                else:
                    yield SSEEvent(data="", event="ping")
                await asyncio.sleep(5)
    """

    def decorator(view_func: Callable) -> Callable:
        # -------------------------------------------------------------------- #
        # Async path                                                           #
        # -------------------------------------------------------------------- #
        if inspect.isasyncgenfunction(view_func):

            @wraps(view_func)
            async def awrapper(
                request: HttpRequest, *args: Any, **kwargs: Any
            ) -> HttpResponse:
                # ---------------------------------------------------------------- #
                # 1. Async streaming generator                                     #
                # ---------------------------------------------------------------- #
                async def astream():
                    if retry is not None:
                        yield f"retry: {retry}\n\n"

                    try:
                        async for item in view_func(request, *args, **kwargs):
                            try:
                                yield to_sse(item)
                            except Exception as fmt_err:  # noqa: BLE001
                                logger.warning("SSE format error: %s", fmt_err)
                                yield SSEEvent(
                                    data=str(fmt_err), event="error"
                                ).encode()

                    except Exception as exc:  # noqa: BLE001
                        logger.exception(
                            "SSE async producer raised an exception: %s", exc
                        )
                        yield SSEEvent(data=str(exc), event="error").encode()

                # ---------------------------------------------------------------- #
                # 2. Build response                                                #
                # ---------------------------------------------------------------- #
                return _build_response(astream())

            return awrapper

        # -------------------------------------------------------------------- #
        # Sync path                                                            #
        # -------------------------------------------------------------------- #
        @wraps(view_func)
        def wrapper(request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
            # ---------------------------------------------------------------- #
            # 1. Streaming generator                                           #
            # ---------------------------------------------------------------- #
            def _stream():
                if retry is not None:
                    yield f"retry: {retry}\n\n"

                try:
                    for item in view_func(request, *args, **kwargs):
                        try:
                            yield to_sse(item)
                        except Exception as fmt_err:  # noqa: BLE001
                            logger.warning("SSE format error: %s", fmt_err)
                            yield SSEEvent(data=str(fmt_err), event="error").encode()

                except GeneratorExit:
                    logger.debug("SSE client disconnected from %s.", request.path)

                except Exception as exc:  # noqa: BLE001
                    logger.exception("SSE producer raised an exception: %s", exc)
                    yield SSEEvent(data=str(exc), event="error").encode()

            # ---------------------------------------------------------------- #
            # 2. Build response                                                #
            # ---------------------------------------------------------------- #
            return _build_response(_stream())

        return wrapper

    # Support both @sse_stream and @sse_stream()
    if _view_func is not None:
        return decorator(_view_func)

    return decorator

SSEEvent

django_flosse.events.SSEEvent dataclass

Represents a single Server-Sent Event.

Attributes: data: The payload. Dicts/lists are serialised to JSON automatically. event: Optional named event type (maps to the event: field). id: Optional event ID for client reconnection (maps to id:). retry: Optional reconnection delay in milliseconds (maps to retry:).

Examples::

SSEEvent(data="hello")
SSEEvent(data={"count": 1}, event="update")
SSEEvent(data="done", event="finish", id="42")
Source code in django_flosse/events.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@dataclass
class SSEEvent:
    """
    Represents a single Server-Sent Event.

    Attributes:
        data:   The payload. Dicts/lists are serialised to JSON automatically.
        event:  Optional named event type (maps to the ``event:`` field).
        id:     Optional event ID for client reconnection (maps to ``id:``).
        retry:  Optional reconnection delay in milliseconds (maps to ``retry:``).

    Examples::

        SSEEvent(data="hello")
        SSEEvent(data={"count": 1}, event="update")
        SSEEvent(data="done", event="finish", id="42")
    """

    data: Any
    event: Optional[str] = None
    id: Optional[str] = None
    retry: Optional[int] = None

    def encode(self) -> str:
        """Serialise to the SSE wire format (ends with a blank line)."""
        lines: list[str] = []

        if self.retry is not None:
            lines.append(f"retry: {self.retry}")

        if self.id is not None:
            lines.append(f"id: {self.id}")

        if self.event is not None:
            lines.append(f"event: {self.event}")

        payload = self.data
        if not isinstance(payload, str):
            payload = json.dumps(payload, ensure_ascii=False)

        # Multi-line data: each line must have its own "data:" prefix.
        for line in payload.splitlines():
            lines.append(f"data: {line}")

        lines.append("")  # Blank line to terminate the event
        return "\n".join(lines) + "\n"

encode() -> str

Serialise to the SSE wire format (ends with a blank line).

Source code in django_flosse/events.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def encode(self) -> str:
    """Serialise to the SSE wire format (ends with a blank line)."""
    lines: list[str] = []

    if self.retry is not None:
        lines.append(f"retry: {self.retry}")

    if self.id is not None:
        lines.append(f"id: {self.id}")

    if self.event is not None:
        lines.append(f"event: {self.event}")

    payload = self.data
    if not isinstance(payload, str):
        payload = json.dumps(payload, ensure_ascii=False)

    # Multi-line data: each line must have its own "data:" prefix.
    for line in payload.splitlines():
        lines.append(f"data: {line}")

    lines.append("")  # Blank line to terminate the event
    return "\n".join(lines) + "\n"

to_sse

django_flosse.formatters.to_sse(raw: Any) -> str

Convert any value yielded by a view into an SSE-formatted string.

Supported yield styles
  • str → treated as plain data
  • (event, data) → named event with data
  • dict → passed as keyword args to SSEEvent; if no "data" key, the whole dict is the data
  • SSEEvent → encoded as-is

Anything else is coerced to str and used as data.

Source code in django_flosse/formatters.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def to_sse(raw: Any) -> str:
    """
    Convert any value yielded by a view into an SSE-formatted string.

    Supported yield styles
    ----------------------
    - ``str``              → treated as plain data
    - ``(event, data)``   → named event with data
    - ``dict``            → passed as keyword args to SSEEvent;
                            if no ``"data"`` key, the whole dict is the data
    - ``SSEEvent``        → encoded as-is

    Anything else is coerced to ``str`` and used as data.
    """

    if isinstance(raw, SSEEvent):
        return raw.encode()

    if isinstance(raw, str):
        return SSEEvent(data=raw).encode()

    if isinstance(raw, tuple):
        if len(raw) == 2:
            event_name, data = raw
            return SSEEvent(event=event_name, data=data).encode()

        if len(raw) == 3:
            event_name, data, id_ = raw
            return SSEEvent(event=event_name, data=data, id=str(id_)).encode()

        raise ValueError(
            f"Tuple yields must be (event, data) or (event, data, id), got {len(raw)}-tuple."
        )

    if isinstance(raw, dict):
        if "data" in raw:
            return SSEEvent(
                **{
                    k: v
                    for k, v in raw.items()
                    if k in ("data", "event", "id", "retry")
                }
            ).encode()

        return SSEEvent(data=raw).encode()

    # Fallback
    return SSEEvent(data=str(raw)).encode()

Exceptions

django_flosse.exceptions.SSEClientDisconnected

Bases: Exception

Raised internally when the client closes the connection.

Source code in django_flosse/exceptions.py 
1
2
class SSEClientDisconnected(Exception):
    """Raised internally when the client closes the connection."""

django_flosse.exceptions.SSEYieldError

Bases: ValueError

Raised when a view yields a value that cannot be converted to an SSE event.

Source code in django_flosse/exceptions.py 
5
6
class SSEYieldError(ValueError):
    """Raised when a view yields a value that cannot be converted to an SSE event."""