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:

modifyvalidate

views.py

Show imports... 7 lines hidden

[source]

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": "Server sent event payload.",
        "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?

1. We define an event producing method produce_user_events yielding events one by one. It returns an collections.abc.AsyncIterator instance

2. It must produce instances of dmr.streaming.sse.metadata.SSEvent, which will be rendered into the stream

3. 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

4. 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

Show imports... 9 lines hidden

[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: Thu, 07 May 2026 12:58:34 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": "Server sent event payload.",
        "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:

1. Cookies based auth, because EventSource passes all the cookies on the request

2. Query string based auth, but it might be exposed in logs / etc, so make sure tokens have a really short expiration time

3. Using your own EventSource

Here’s an example with DjangoSessionAsyncAuth class:

views.py

Show imports... 8 lines hidden

[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": "Server sent event payload.",
        "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:

1. 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

2. If any new payment is made, send us payment event type with {"amount": int, "currency": str} json string as the data

3. 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.

Show imports... 11 lines hidden

[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)

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

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

Return a ping event’s payload.

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

Send ping keep alive events every 15 seconds.

classmethod streaming_renderers(serializer: type[_SerializerT_co]) → list[StreamingRenderer]

Returns the iterable of streaming renderers for this controller.

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)

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)

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)

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

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

Server sent event payload.

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'))

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

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)

Bases: StreamingValidator

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

validation_pipeline() → Iterable[Callable[[SSE, Any, type[BaseSerializer]], SSE]]

Validate the event type and the event payload.

dmr.streaming.sse.validation.validate_event_data(event: Any, model: Any, serializer: type[BaseSerializer]) → Any

Validates SSEvent.data to be of the given type arg.

dmr.streaming.sse.validation.check_event_field(event_field: Any, field_name: str) → None

Checks that event field does not contain wrong chars.

Exceptions

final exception dmr.streaming.exceptions.StreamingCloseError

Raised when we need to immediately close the response stream.

Raise it from events producing async iterator.