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"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/usereventscontroller/": {
      "get": {
        "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 produce instances of objects that can be serialized to json with a serializer of your choice. These events will be rendered 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: Fri, 16 Oct 2026 22:54:41 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"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/usereventscontroller/": {
      "get": {
        "operationId": "getUserEventsControllerApiUsereventscontroller",
        "parameters": [
          {
            "in": "header",
            "name": "Last-Event-ID",
            "schema": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "null"
                }
              ],
              "default": 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 HeaderJWTAsyncAuth 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 HeaderJWTAsyncAuth
from dmr.streaming.jsonl import JsonLinesController


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


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

    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 -H 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNzkyMjc3NjgyLCJpYXQiOjE3OTIxOTEyODIsImp0aSI6IjIxNThiNDI4ZmE3MTQ3NWFhOWNmN2VhY2ZhNmJjYjE4IiwiZXh0cmFzIjp7InR5cGUiOiJhY2Nlc3MifX0.PRvE818TJACln6a1Nu6yf4GBF5WnuJuwcXhL1Fnio1E'
{"email":"first@example.com"}

$ 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": {
        "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",
            "headers": {
              "WWW-Authenticate": {
                "description": "Challenges that the client can use to authenticate this request",
                "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"
          }
        },
        "security": [
          {
            "jwt": []
          }
        ]
      }
    }
  }
}

See also

Read our How authentication works guide.

Best practices

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

• 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 setting ping_seconds in Streaming extras to the maximum number of seconds before the ping event happens, either per controller with extras = Streaming(ping_seconds=30) or per endpoint with @modify(extras=Streaming(ping_seconds=30)). And ping_event() for the event payload.

See API reference.