Server Sent Events¶

Standard: https://html.spec.whatwg.org/multipage/server-sent-events.html

Our SSE implementation allows users to follow the standard above or fully customize the experience for custom needs.

Using SSE¶

You can use SSE with both validate() and modify() style endpoints:

views.py¶

import dataclasses
from collections.abc import AsyncIterator

from dmr.plugins.msgspec import MsgspecSerializer
from dmr.streaming.sse import SSEController, SSEvent


@dataclasses.dataclass(frozen=True, slots=True)
class _User:
    email: str


class UserEventsController(SSEController[MsgspecSerializer]):
    async def get(self) -> AsyncIterator[SSEvent[_User]]:
        return self.produce_user_events()

    async def produce_user_events(self) -> AsyncIterator[SSEvent[_User]]:
        # You can send any complex data that can be serialized
        # by the controller's serializer,
        # all SSEvent fields can be customized:
        yield SSEvent(
            _User(email='first@example.com'),
            event='user',
        )
        yield SSEvent(
            _User(email='second@example.com'),
            event='user',
        )

Run result

$ curl http://127.0.0.1:8000/api/user/events/ -X GET
event: user
data: {"email":"first@example.com"}

event: user
data: {"email":"second@example.com"}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "type": "array"
          },
          "msg": {
            "type": "string"
          },
          "type": {
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      },
      "SSEvent__User_": {
        "description": "Default implementation for the Server Sent Event.\n\nAll parameters are optional, but at least one is required.\n\nAttributes:\n    data: Event payload.\n    event: Event type.\n    id: Unique event's identification.\n    retry: The reconnection time.\n    comment: Comment about the event.\n\n.. note::\n\n    It is recommended for end users to define their own types\n    that will be type-safe and will have the correct schema.\n\nSee also:\n    https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#fields",
        "properties": {
          "comment": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "data": {
            "$ref": "#/components/schemas/_User"
          },
          "event": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "id": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "retry": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ]
          }
        },
        "required": [
          "data"
        ],
        "title": "SSEvent[_User]",
        "type": "object"
      },
      "_User": {
        "properties": {
          "email": {
            "type": "string"
          }
        },
        "required": [
          "email"
        ],
        "title": "_User",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/usereventscontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getUsereventscontrollerApiUsereventscontroller",
        "responses": {
          "200": {
            "content": {
              "text/event-stream": {
                "itemSchema": {
                  "$ref": "#/components/schemas/SSEvent__User_"
                }
              }
            },
            "description": "OK",
            "headers": {
              "Cache-Control": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "Connection": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-Accel-Buffering": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          }
        }
      }
    }
  }
}

views.py¶

[source]

import dataclasses
from collections.abc import AsyncIterator

from dmr import validate
from dmr.negotiation import ContentType
from dmr.plugins.msgspec import MsgspecSerializer
from dmr.streaming import StreamingResponse, streaming_response_spec
from dmr.streaming.sse import SSEController, SSEvent


@dataclasses.dataclass(frozen=True, slots=True)
class _User:
    email: str


class UserEventsController(SSEController[MsgspecSerializer]):
    @validate(
        streaming_response_spec(
            SSEvent[_User],
            content_type=ContentType.event_stream,
        ),
    )
    async def get(self) -> StreamingResponse:
        return self.to_stream(self.produce_user_events())

    async def produce_user_events(self) -> AsyncIterator[SSEvent[_User]]:
        # You can send any complex data that can be serialized
        # by the controller's serializer,
        # all SSEvent fields can be customized:
        yield SSEvent(
            _User(email='first@example.com'),
            event='user',
        )
        yield SSEvent(
            _User(email='second@example.com'),
            event='user',
        )

Run result

$ curl http://127.0.0.1:8000/api/user/events/ -X GET
event: user
data: {"email":"first@example.com"}

event: user
data: {"email":"second@example.com"}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "type": "array"
          },
          "msg": {
            "type": "string"
          },
          "type": {
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      },
      "SSEvent__User_": {
        "description": "Default implementation for the Server Sent Event.\n\nAll parameters are optional, but at least one is required.\n\nAttributes:\n    data: Event payload.\n    event: Event type.\n    id: Unique event's identification.\n    retry: The reconnection time.\n    comment: Comment about the event.\n\n.. note::\n\n    It is recommended for end users to define their own types\n    that will be type-safe and will have the correct schema.\n\nSee also:\n    https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#fields",
        "properties": {
          "comment": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "data": {
            "$ref": "#/components/schemas/_User"
          },
          "event": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "id": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "retry": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ]
          }
        },
        "required": [
          "data"
        ],
        "title": "SSEvent[_User]",
        "type": "object"
      },
      "_User": {
        "properties": {
          "email": {
            "type": "string"
          }
        },
        "required": [
          "email"
        ],
        "title": "_User",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/usereventscontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getUsereventscontrollerApiUsereventscontroller",
        "responses": {
          "200": {
            "content": {
              "text/event-stream": {
                "itemSchema": {
                  "$ref": "#/components/schemas/SSEvent__User_"
                }
              }
            },
            "description": "OK",
            "headers": {
              "Cache-Control": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "Connection": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-Accel-Buffering": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          }
        }
      }
    }
  }
}

What happens in these examples?

We define an event producing method produce_user_events yielding events one by one. It returns an collections.abc.AsyncIterator instance
It must produces instances of dmr.streaming.sse.metadata.SSEvent, which will be renderer into the stream
We define a special SSEController class that has regular get HTTP endpoint. In @modify example it returns the async generator directly, while in @validate example it returns the dmr.streaming.stream.StreamingResponse instance
Next, ASGI will take the returned data and stream events to your users

See also

Async Generators in Python: https://peps.python.org/pep-0525
Streaming in ASGI: https://asgi.readthedocs.io/en/latest/specs/main.html

Important

Our streaming implementation will not work with a WSGI handler in production. Why? Because streaming is a long-living connection by design. WSGI handlers have very limited number of connections. Basically number_of_workers * number_of_threads, just a very small number of streaming clients will completely block all other work on the server.

Use ASGI for all streaming endpoints. This will give you the best of two worlds: simple sync Django for the major part of your code base and some async endpoints where you need them. See our guide.

However, we allow running streaming with WSGI if settings.DEBUG is True for local development and testing. In a very limited compatibility mode.

Using components¶

If you want to parse any incoming data, you can do it the same way as in any other controller.

Note

Note that default EventSource JavaScript API only support headers, cookies, query, and path parameters in GET HTTP method.

Custom implementations might use any HTTP methods and any type of parameters.

For example, if you need to parse Last-Event-ID header (which is a part of the default EventSource spec and API):

views.py¶

[source]

from collections.abc import AsyncIterator

import msgspec

from dmr.components import Headers
from dmr.plugins.msgspec import MsgspecSerializer
from dmr.streaming.sse import SSEController, SSEvent


class HeaderModel(msgspec.Struct):
    last_event_id: int | None = msgspec.field(
        default=None,
        name='Last-Event-ID',
    )


class UserEventsController(SSEController[MsgspecSerializer]):
    def get(
        self,
        parsed_headers: Headers[HeaderModel],
    ) -> AsyncIterator[SSEvent[str]]:
        return self.produce_user_events(parsed_headers)

    async def produce_user_events(
        self,
        parsed_headers: HeaderModel,
    ) -> AsyncIterator[SSEvent[str]]:
        if parsed_headers.last_event_id is None:
            yield SSEvent('starting from scratch')
        else:
            yield SSEvent(f'starting from {parsed_headers.last_event_id}')

Run result

$ curl http://127.0.0.1:8000/api/usereventscontroller/ -X GET
data: "starting from scratch"


$ curl http://127.0.0.1:8000/api/usereventscontroller/ -X GET -H 'Last-Event-ID: 5'
data: "starting from 5"


$ curl http://127.0.0.1:8000/api/usereventscontroller/ -D - -X GET -H 'Last-Event-ID: abc'
HTTP/1.1 400 Bad Request
date: Sun, 29 Mar 2026 18:53:19 GMT
server: uvicorn
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 114
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

{"detail":[{"msg":"Expected `int | null`, got `str` - at `$.parsed_headers.last_event_id`","type":"value_error"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "type": "array"
          },
          "msg": {
            "type": "string"
          },
          "type": {
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      },
      "SSEvent_str_": {
        "description": "Default implementation for the Server Sent Event.\n\nAll parameters are optional, but at least one is required.\n\nAttributes:\n    data: Event payload.\n    event: Event type.\n    id: Unique event's identification.\n    retry: The reconnection time.\n    comment: Comment about the event.\n\n.. note::\n\n    It is recommended for end users to define their own types\n    that will be type-safe and will have the correct schema.\n\nSee also:\n    https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#fields",
        "properties": {
          "comment": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "data": {
            "type": "string"
          },
          "event": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "id": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "retry": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ]
          }
        },
        "required": [
          "data"
        ],
        "title": "SSEvent[str]",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/usereventscontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getUsereventscontrollerApiUsereventscontroller",
        "parameters": [
          {
            "deprecated": false,
            "in": "header",
            "name": "Last-Event-ID",
            "schema": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "null"
                }
              ]
            }
          }
        ],
        "responses": {
          "200": {
            "content": {
              "text/event-stream": {
                "itemSchema": {
                  "$ref": "#/components/schemas/SSEvent_str_"
                }
              }
            },
            "description": "OK",
            "headers": {
              "Cache-Control": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "Connection": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-Accel-Buffering": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "400": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when request components cannot be parsed"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          }
        }
      }
    }
  }
}

We are using a regular approach with the Headers component.

Note

Use Last-Event-ID header to handle reconnects to start sending events to the client from the last consumed one.

See also

Read our Components guide.

Auth¶

SSE endpoints can also be protected by any instance of the async auth. However, note that default EventSource JavaScript API does not support passing explicit headers. There are several options:

Cookies based auth, because EventSource passes all the cookies on the request
Query string based auth, but it might be exposed in logs / etc, so make sure tokens have a really short expiration time
Using your own EventSource

Here’s an example with DjangoSessionAsyncAuth class:

views.py¶

[source]

import dataclasses
from collections.abc import AsyncIterator

from dmr.plugins.msgspec import MsgspecSerializer
from dmr.security.django_session import DjangoSessionAsyncAuth
from dmr.streaming.sse import SSEController, SSEvent


@dataclasses.dataclass(frozen=True, slots=True)
class _User:
    email: str


class UserEventsController(SSEController[MsgspecSerializer]):
    auth = (DjangoSessionAsyncAuth(),)

    async def get(self) -> AsyncIterator[SSEvent[_User]]:
        return self.produce_user_events()

    async def produce_user_events(self) -> AsyncIterator[SSEvent[_User]]:
        yield SSEvent(
            _User(email='first@example.com'),
            event='user',
        )

Run result

$ curl http://127.0.0.1:8000/api/user/events/ -X GET
{"detail":[{"msg":"Not authenticated","type":"security"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "type": "array"
          },
          "msg": {
            "type": "string"
          },
          "type": {
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      },
      "SSEvent__User_": {
        "description": "Default implementation for the Server Sent Event.\n\nAll parameters are optional, but at least one is required.\n\nAttributes:\n    data: Event payload.\n    event: Event type.\n    id: Unique event's identification.\n    retry: The reconnection time.\n    comment: Comment about the event.\n\n.. note::\n\n    It is recommended for end users to define their own types\n    that will be type-safe and will have the correct schema.\n\nSee also:\n    https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#fields",
        "properties": {
          "comment": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "data": {
            "$ref": "#/components/schemas/_User"
          },
          "event": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "id": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ]
          },
          "retry": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ]
          }
        },
        "required": [
          "data"
        ],
        "title": "SSEvent[_User]",
        "type": "object"
      },
      "_User": {
        "properties": {
          "email": {
            "type": "string"
          }
        },
        "required": [
          "email"
        ],
        "title": "_User",
        "type": "object"
      }
    },
    "securitySchemes": {
      "csrf": {
        "description": "CSRF protection",
        "in": "cookie",
        "name": "csrftoken",
        "type": "apiKey"
      },
      "django_session": {
        "description": "Reusing standard Django auth flow for API",
        "in": "cookie",
        "name": "sessionid",
        "type": "apiKey"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/usereventscontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getUsereventscontrollerApiUsereventscontroller",
        "responses": {
          "200": {
            "content": {
              "text/event-stream": {
                "itemSchema": {
                  "$ref": "#/components/schemas/SSEvent__User_"
                }
              }
            },
            "description": "OK",
            "headers": {
              "Cache-Control": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "Connection": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-Accel-Buffering": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "401": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when auth was not successful"
          },
          "403": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when CSRF check failed"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          }
        },
        "security": [
          {
            "csrf": [],
            "django_session": []
          }
        ]
      }
    }
  }
}

If you don’t use EventSource API, you can use any other auth of your choice, all of them will just work.

See also

Read our How authentication works guide.

Serializing events¶

Our default class SSEvent supports two modes:

Passing serialize=True (default) for all event bodies, so they will be serialized with the serializer from the controller. In this mode you can pass any objects that are supported by your serializer
Or passing serialize=False alongside the existing bytes object. It won’t trigger any extra serialization. It might be useful if you already have some existing binary data

Modeling business events¶

We provide our default implementation for sending events: SSEvent

But, users are not required to use it directly. They can create their own models, as long as they respect SSE protocol fields.

It is quite common in SSE to model different ADTs. Because events can be of different types, they might have different data based on it. And they might also contain different other fields based on that.

For example, let’s model three different events:

If any new users are registered, send us an event with type user, id with the user’s id, and a username as the data
If any new payment is made, send us payment event type with {"amount": int, "currency": str} json string as the data
Sometimes we send purely technical ping events with : ping as a comment and retry: 50 instruction

Let’s model this with perfect type-safety and state-of-the-art OpenAPI schema.

[source]

from collections.abc import AsyncIterator
from typing import Literal, TypeAlias

import pydantic
from pydantic.json_schema import SkipJsonSchema

from dmr.errors import ErrorModel
from dmr.plugins.pydantic import PydanticSerializer
from dmr.streaming.sse import SSEController


class _BaseEvent(pydantic.BaseModel):
    comment: SkipJsonSchema[str | None] = None
    retry: SkipJsonSchema[int | None] = None

    @property
    def should_serialize_data(self) -> bool:
        return True


class UserEvent(_BaseEvent):
    event: Literal['user'] = 'user'
    id: int
    data: str  # username


class _Payment(pydantic.BaseModel):
    amount: int
    currency: str


class PaymentEvent(_BaseEvent):
    id: SkipJsonSchema[None] = None
    event: Literal['payment'] = 'payment'
    data: pydantic.Json[_Payment]


class PingEvent(pydantic.BaseModel):
    retry: int = 100
    comment: Literal['ping'] = 'ping'
    id: SkipJsonSchema[None] = None
    data: SkipJsonSchema[None] = None
    event: SkipJsonSchema[None] = None

    @property
    def should_serialize_data(self) -> bool:
        return False


class ErrorEvent(_BaseEvent):
    # Can happen if event validation will fail:
    id: SkipJsonSchema[None] = None
    event: Literal['error'] = 'error'
    data: pydantic.Json[ErrorModel]


_PossibleEvents: TypeAlias = UserEvent | PaymentEvent | PingEvent | ErrorEvent


class ComplexEventsController(SSEController[PydanticSerializer]):
    def get(self) -> AsyncIterator[_PossibleEvents]:
        return self.complex_events()

    async def complex_events(self) -> AsyncIterator[_PossibleEvents]:
        yield UserEvent(id=1, data='sobolevn')
        yield PaymentEvent(
            data=_Payment(
                amount=10,
                currency='$',
            ).model_dump_json(),
        )
        yield PingEvent()

Run result

$ curl http://127.0.0.1:8000/api/complexeventscontroller/ -X GET
id: 1
event: user
data: "sobolevn"

event: payment
data: {"amount":10,"currency":"$"}

: ping
retry: 100

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorEvent": {
        "properties": {
          "data": {
            "contentMediaType": "application/json",
            "contentSchema": {
              "$ref": "#/components/schemas/ErrorModel"
            },
            "title": "Data",
            "type": "string"
          },
          "event": {
            "const": "error",
            "default": "error",
            "title": "Event",
            "type": "string"
          }
        },
        "required": [
          "data"
        ],
        "title": "ErrorEvent",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      },
      "PaymentEvent": {
        "properties": {
          "data": {
            "contentMediaType": "application/json",
            "contentSchema": {
              "$ref": "#/components/schemas/_Payment"
            },
            "title": "Data",
            "type": "string"
          },
          "event": {
            "const": "payment",
            "default": "payment",
            "title": "Event",
            "type": "string"
          }
        },
        "required": [
          "data"
        ],
        "title": "PaymentEvent",
        "type": "object"
      },
      "PingEvent": {
        "properties": {
          "comment": {
            "const": "ping",
            "default": "ping",
            "title": "Comment",
            "type": "string"
          },
          "retry": {
            "default": 100,
            "title": "Retry",
            "type": "integer"
          }
        },
        "title": "PingEvent",
        "type": "object"
      },
      "UserEvent": {
        "properties": {
          "data": {
            "title": "Data",
            "type": "string"
          },
          "event": {
            "const": "user",
            "default": "user",
            "title": "Event",
            "type": "string"
          },
          "id": {
            "title": "Id",
            "type": "integer"
          }
        },
        "required": [
          "id",
          "data"
        ],
        "title": "UserEvent",
        "type": "object"
      },
      "_Payment": {
        "properties": {
          "amount": {
            "title": "Amount",
            "type": "integer"
          },
          "currency": {
            "title": "Currency",
            "type": "string"
          }
        },
        "required": [
          "amount",
          "currency"
        ],
        "title": "_Payment",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/complexeventscontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getComplexeventscontrollerApiComplexeventscontroller",
        "responses": {
          "200": {
            "content": {
              "text/event-stream": {
                "itemSchema": {
                  "anyOf": [
                    {
                      "$ref": "#/components/schemas/UserEvent"
                    },
                    {
                      "$ref": "#/components/schemas/PaymentEvent"
                    },
                    {
                      "$ref": "#/components/schemas/PingEvent"
                    },
                    {
                      "$ref": "#/components/schemas/ErrorEvent"
                    }
                  ]
                }
              }
            },
            "description": "OK",
            "headers": {
              "Cache-Control": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "Connection": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-Accel-Buffering": {
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          }
        }
      }
    }
  }
}

This will also generate a correct OpenAPI spec with all the logical cases covered.

If you are still not happy with the resulting OpenAPI schema, you can fully customize it using your serializer’s official docs. For example, pydantic uses __get_pydantic_json_schema__ method for this purpose.

Note

When creating custom event types, don’t forget to validate that id and event fields do not contain: '\x00', '\n', and '\r' chars.

Use dmr.streaming.sse.validation.check_event_field() to do that.

Best practices¶

django-modern-rest implements a bunch of best practices for streaming SSE:

Connection: keep-alive header keeps the connection open
Cache-Control: no-cache header prevents caching the stream response
X-Accel-Buffering: no header prevents proxy response buffering in some proxy servers like Nginx
Every 15 seconds we send : ping keep-alive events, when there hasn’t been any message, to prevent some servers from closing the connection as inactive. This is a direct recommendation from the SSE spec

Everything just works out of the box, you don’t have to do anything.

API Reference¶

Controller¶

class dmr.streaming.sse.controller.SSEController(**kwargs)[source]¶

Bases: StreamingController[_SerializerT_co]

Controller for streaming Server Sent Events (SSE).

Danger

WSGI handers do not support streaming responses, including SSE, by default. You would need to use ASGI handler for streaming endpoints.

We allow running SSE during settings.DEBUG builds for debugging. But, in production we will raise RuntimeError when WSGI handler will be detected used together with streaming.

async handle_event_error(exc: Exception) → Any[source]¶

Handles errors that can happen while sending events.

Return alternative event that will indicate what error has happened. By default does nothing and just reraises the exception.

ping_event() → Any | None[source]¶: Return a ping event to be generated if this streaming needs it.

streaming_default_renderer: ClassVar[Renderer] = <dmr.plugins.msgspec.json.MsgspecJsonRenderer object>¶: Default renderer for event body field.

streaming_ping_seconds: ClassVar[float | None] = 15.0¶

Optional ping keep alive event support.

Some servers might close long living connections with no activity. Specify number in second how long should we wait between events. If we wait longer, we will send a ping event. The payload of the ping event is defined in ping_event().

By default it is disabled. It is only enabled in the SSE streaming.

classmethod streaming_renderers(serializer: type[_SerializerT_co]) → list[StreamingRenderer][source]¶: Returns the streaming renderer.

streaming_validator_cls¶

Validator for events, only active when validate_events is set.

alias of SSEStreamingValidator

Metadata¶

class dmr.streaming.sse.metadata.SSE(*args, **kwargs)[source]¶

Basic interface for all possible SSE implementations.

We don’t force users to use our default implementation, moreover, we encourage them to create their own event ADT and models.

property should_serialize_data: bool¶: Should we serialize the data attribute or return it as is?

final class dmr.streaming.sse.metadata.SSEvent(data: None = None, *, event: str, id: int | str | None = None, retry: int | None = None, comment: str | None = None, serialize: bool = True)[source]¶

final class dmr.streaming.sse.metadata.SSEvent(data: None = None, *, event: str | None = None, id: int | str, retry: int | None = None, comment: str | None = None, serialize: bool = True)

final class dmr.streaming.sse.metadata.SSEvent(data: None = None, *, event: str | None = None, id: int | str | None = None, retry: int, comment: str | None = None, serialize: bool = True)

final class dmr.streaming.sse.metadata.SSEvent(data: None = None, *, event: str | None = None, id: int | str | None = None, retry: int | None = None, comment: str, serialize: bool = True)

Default implementation for the Server Sent Event.

All parameters are optional, but at least one is required.

data¶

Event payload.

Type:: dmr.streaming.sse.metadata._DataT_co

event¶

Event type.

Type:: str | None

id¶

Unique event’s identification.

Type:: int | str | None

retry¶

The reconnection time.

Type:: int | None

comment¶

Comment about the event.

Type:: str | None

Note

It is recommended for end users to define their own types that will be type-safe and will have the correct schema.

property should_serialize_data: bool¶

Should we serialize data attribute with the serializer?

Serializes by default. When serialize is False, data can only be bytes.

Renderer¶

class dmr.streaming.sse.renderer.SSERenderer(serializer: type[BaseSerializer], regular_renderer: Renderer, streaming_validator_cls: type[StreamingValidator], *, sep: bytes = b'\r\n', encoding: str = 'utf-8', linebreak: Pattern[bytes] = re.compile(b'\\r\\n|\\r|\\n'))[source]¶

Renders response as a stream of SSE.

Uses sub-renderer to render events’ data into the correct format.

content_type: str = 'text/event-stream'¶

Content-Type that this renderer works with.

Must be defined for all subclasses.

render(to_serialize: Any, serializer_hook: Callable[[Any], Any]) → bytes[source]¶: Render a single event in the SSE stream of events.

Validation¶

class dmr.streaming.sse.validation.SSEStreamingValidator(event_model: Any, serializer: type[BaseSerializer], *, validate_events: bool)[source]¶

Bases: StreamingValidator

Injects itself into the stream of SSE to validate the events.

validation_pipeline() → Iterable[Callable[[SSE, Any, type[BaseSerializer]], SSE]][source]¶: Validate the event type and the event payload.

dmr.streaming.sse.validation.validate_event_data(event: Any, model: Any, serializer: type[BaseSerializer]) → Any[source]¶: Validates SSEvent.data to be of the given type arg.

dmr.streaming.sse.validation.check_event_field(event_field: Any, field_name: str) → None[source]¶: Checks that event field does not contain wrong chars.

Exceptions¶

final exception dmr.streaming.exceptions.StreamingCloseError[source]¶

Raised when we need to imediatelly close the response stream.

Raise it from events producing async iterator.