Json Lines

Standard: https://jsonlines.org

Our jsonl implementation allows users to follow the standard above.

Using JsonLines

You can use JsonLines format 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.jsonl import JsonLinesController


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


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

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

Run result

$ curl http://127.0.0.1:8000/api/user/events/ -X GET
{"email":"first@example.com"}
{"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"
      },
      "_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": {
              "application/jsonl": {
                "itemSchema": {
                  "$ref": "#/components/schemas/_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 produces instances of objects that can be serialized to json with a serializer of your choice. These events will be renderer into a stream

3. We define a special JsonLinesController 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.

JsonL supports passing any type of data to the endpoint.

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.jsonl import Json, JsonLinesController


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


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

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

Run result

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

$ curl http://127.0.0.1:8000/api/usereventscontroller/ -X GET -H 'Last-Event-ID: 5'
"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:16 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"
      }
    },
    "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": {
              "application/jsonl": {
                "itemSchema": {}
              }
            },
            "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

JsonL endpoints fully support any style of auth that you might need.

Here’s an example with JWTAsyncAuth 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.jwt import JWTAsyncAuth
from dmr.streaming.jsonl import JsonLinesController


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


class UserEventsController(JsonLinesController[MsgspecSerializer]):
    auth = (JWTAsyncAuth(),)

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

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

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"
      },
      "_User": {
        "properties": {
          "email": {
            "type": "string"
          }
        },
        "required": [
          "email"
        ],
        "title": "_User",
        "type": "object"
      }
    },
    "securitySchemes": {
      "jwt": {
        "bearerFormat": "JWT",
        "description": "JWT token auth",
        "scheme": "Bearer",
        "type": "http"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/usereventscontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getUsereventscontrollerApiUsereventscontroller",
        "responses": {
          "200": {
            "content": {
              "application/jsonl": {
                "itemSchema": {
                  "$ref": "#/components/schemas/_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"
          },
          "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": [
          {
            "jwt": []
          }
        ]
      }
    }
  }
}

See also

Read our How authentication works guide.

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

Everything just works out of the box, you don’t have to do anything. However, we don’t send ping events by default, because the format for them is not well defined in jsonl.

You can enable them by changing streaming_ping_seconds to the maximum number of second before the ping event happens. And ping_event() for the event payload.

API Reference

Controller

class dmr.streaming.jsonl.controller.JsonLinesController(**kwargs)

Bases: StreamingController[_SerializerT_co]

Controller for streaming json lines (JsonL).

See also

Json Lines standard: https://jsonlines.org

Danger

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

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

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.

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

Returns the streaming renderer.

streaming_validator_cls

alias of JsonLinesStreamingValidator

Renderer

class dmr.streaming.jsonl.renderer.JsonLinesRenderer(serializer: type[BaseSerializer], regular_renderer: Renderer, streaming_validator_cls: type[StreamingValidator], *, sep: bytes = b'\n')

Renders response as a stream of json liens.

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

content_type: str = 'application/jsonl'

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 json lines stream of events.

Validation

class dmr.streaming.jsonl.validation.JsonLinesStreamingValidator(event_model: Any, serializer: type[BaseSerializer], *, validate_events: bool)

Bases: StreamingValidator

Injects itself into the stream of json lines to validate the events.

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

Validate the event type and the event payload.