Streaming

What is a stream?

Streaming is a bit different from regular REST APIs. REST APIs always return the whole data in their response.

Streaming establishes a persistent connection, accepts headers with the content type of application/jsonl (stands for “JSON Lines”) or text/event-stream and then streams each individual event when it is ready one by one.

        ---
config:
  theme: forest

---
sequenceDiagram
    participant App
    participant Client

    Client->>App: Make the initial request
    App->>Client: Establish connection and send headers

    App->>App: Produce Event 1
    App->>Client: Send Event 1
    Client->>Client: Process Event 1

    App->>App: Produce Event 2
    App->>Client: Send Event 2
    Client->>Client: Process Event 2

    Note over App: Continue producing events...
    Note over Client: Continue consuming events...

    Client->>App: Closes the connection
    App->>App: Cleans everything up
    

How streaming works

When to use streaming? When you have a single-directional stream of events.

For example:

• LLM responses

• Logs

• Telemetry

• Financial services

• Sporting events

• Live locations

Do not use streaming for regular responses with regular data. Only use it when needed.

Just a regular controller

We base our API around dmr.streaming.controller.StreamingController type which is a slightly modified subclass of a regular Controller.

Streaming controllers support all the same features:

• Different HTTP async or sync methods

• Components parsing

• How authentication works

• Error handling (including special error handling for events)

• Content negotiation

• Optional Response validation (including events validation)

• etc

We utilize collections.abc.AsyncIterator protocol to model async event sources.

Existing streaming formats

We support:

JsonLinesSSE

JsonLines is a simple format, where events are streamed line by line, and all lines are valid JSON objects.

Read more: Json Lines

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 create our controller as usual, except we are using dmr.streaming.controller.StreamingController subtypes instead of a regular Controller class

2. We return typing.AsyncIterator instance instead of a single data item

Choose the one that fits your needs, or create your own format!

Important

All streaming modes require Django to run in ASGI mode in production.

OpenAPI

We provide full OpenAPI schema support for streaming endpoints. We also support itemSchema schemas for individual events, when OpenAPI 3.2.0 version is used.

Right now, the default version is 3.1.0, so you have to change the version number manually:

views.py

Show imports... 2 lines hidden

[source]

from dmr.openapi import OpenAPIConfig

config = OpenAPIConfig(
    title='My awesome API with streaming',
    version='0.1.0',
    openapi_version='3.2.0',  # NOTE: required to get `itemSchema` support
)

If you are using streaming, using 3.2.0 is a recommended way. In 3.1.0 we will produce regular schema definition, which is not the best way to describe events.

Handling errors

Streaming controllers have two layers of error handling:

1. Default error handling works the same way for request / response phase, see Error handling for more information

2. Per-event error handling, which starts when the streaming connection is established and the events are produced. Such errors are handled in handle_event_error() method

The second layer of event error handling is unique for streaming controllers.

It works for several cases:

1. When event validation fails (mostly useful in development)

2. When there’s an error in the event producer

Let’s see how it can be customized.

JsonLinesSSE

views.py

Show imports... 9 lines hidden

[source]

from collections.abc import AsyncIterator
from typing import Any

from typing_extensions import override

from dmr.plugins.pydantic import PydanticSerializer
from dmr.streaming.jsonl import Json, JsonLinesController


class NumberEventsController(JsonLinesController[PydanticSerializer]):
    async def get(self) -> AsyncIterator[Json]:
        return self._valid_events()

    @override
    async def handle_event_error(self, exc: Exception) -> Any:
        if isinstance(exc, ZeroDivisionError):
            return {'error': 'zero divizion'}
        return await super().handle_event_error(exc)

    async def _valid_events(self) -> AsyncIterator[Json]:
        yield 1
        # Error here:
        yield 1 / 0  # noqa: WPS344  # pyrefly: ignore[division-by-zero]
        # Won't be sent:
        yield 2

Run result

$ curl http://127.0.0.1:8000/api/numbereventscontroller/ -X GET
1
{"error":"zero divizion"}

Warning

Please, note that new events won’t be produced if the error happens in the async generator itself. If you want to handle errors there as well, use try/except right inside the event producing async generator.

Handling disconnects

If you need to immediately close the response stream, you can raise StreamingCloseError or asyncio.CancelledError inside the events producing async iterator.

Async clients can disconnect at any time. We always handle this error gracefully.

See also

Response disconnect docs: https://docs.djangoproject.com/en/stable/ref/request-response/#request-response-streaming-disconnect

Validation

All our regular response validation rules are applied to the SSE controllers as well. We strictly validate that all headers / cookies / etc are listed in the endpoint’s metadata.

We follow the regular rules for response validation and it can be disabled by setting validate_responses=False on the needed level.

We also validate events structure, when streaming them to end users. It is recommended to be turned on in development and turned off in production.

Rules:

• If endpoint specifies validate_events in its Streaming extras, we use it

• If endpoint does not specify this flag, but controller does, we use it

• If controller does not specify this flag, but settings does, we use it

• If no explicit validate_events boolean value is specified, we fallback to validate_responses value

To set validate_events per endpoint, pass Streaming as extras= to the streaming versions of the decorators: modify and validate. Regular modify and validate do not accept streaming extras. If you don’t set validate_events for your endpoints, you are free to use any of these two decorators. They work identically, only typing is different.

per endpointper controllerper settings

Both validate and modify support this extra:

views.py

Show imports... 8 lines hidden

[source]

from collections.abc import AsyncIterator
from typing import Any

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


class UserEventsController(SSEController[MsgspecSerializer]):
    @modify(extras=Streaming(validate_events=False))
    async def get(self) -> AsyncIterator[SSEvent[int]]:
        return self.produce_user_events()

    async def produce_user_events(self) -> AsyncIterator[SSEvent[Any]]:
        yield SSEvent('not-an-int')

Run result

$ curl http://127.0.0.1:8000/api/user/events/ -X GET
data: "not-an-int"

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_int_": {
        "description": "Server sent event payload.",
        "properties": {
          "comment": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "data": {
            "type": "integer"
          },
          "event": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "id": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          },
          "retry": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ],
            "default": null
          }
        },
        "required": [
          "data"
        ],
        "title": "SSEvent[int]",
        "type": "object"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/usereventscontroller/": {
      "get": {
        "operationId": "getUserEventsControllerApiUsereventscontroller",
        "responses": {
          "200": {
            "content": {
              "text/event-stream": {
                "itemSchema": {
                  "$ref": "#/components/schemas/SSEvent_int_"
                }
              }
            },
            "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"
          }
        }
      }
    }
  }
}

How do we know the model for events to be validated against?

• It might be specified as the return_type in the ResponseSpec of @validate for the given status code

• It might be specified as the type argument to generic collections.abc.AsyncIterator return type in @modify styled endpoint.

We fallback to typing.Any if we can’t infer the event model.

Further reading

See also

Best practices for working with async iterators and generators in Python: https://docs.python.org/3.15/library/asyncio-dev.html#asynchronous-generators-best-practices

JsonLines

Streaming individual lines of json.

Json Lines

SSE

Streaming Server Sent Events.

Server Sent Events

Advanced topics

Negotiation for different formats

If you need different event formats for different users / usecases, then you can use the regular Content negotiation process.

Send different Accept headers to get different event streams back:

views.py

Show imports... 19 lines hidden

[source]

from collections.abc import AsyncGenerator, AsyncIterator
from contextlib import aclosing
from typing import Annotated

from typing_extensions import override

from dmr.negotiation import ContentType, conditional_type
from dmr.plugins.pydantic import PydanticSerializer
from dmr.settings import default_renderer
from dmr.streaming import StreamingController
from dmr.streaming.jsonl import Json
from dmr.streaming.jsonl.renderer import JsonLinesRenderer
from dmr.streaming.jsonl.validation import JsonLinesStreamingValidator
from dmr.streaming.renderer import StreamingRenderer
from dmr.streaming.sse import SSEvent
from dmr.streaming.sse.renderer import SSERenderer
from dmr.streaming.sse.validation import SSEStreamingValidator


class EventStreaming(StreamingController[PydanticSerializer]):
    @classmethod
    @override
    def streaming_renderers(
        cls,
        serializer: type[PydanticSerializer],
    ) -> list[StreamingRenderer]:
        return [
            SSERenderer(
                serializer,
                default_renderer,
                SSEStreamingValidator,
            ),
            JsonLinesRenderer(
                serializer,
                default_renderer,
                JsonLinesStreamingValidator,
            ),
        ]

    async def get(
        self,
    ) -> AsyncIterator[
        Annotated[
            SSEvent[Json] | Json,
            conditional_type({
                ContentType.jsonl: Json,
                ContentType.event_stream: SSEvent[Json],
            }),
        ],
    ]:
        return self._valid_events()

    async def _valid_events(self) -> AsyncIterator[SSEvent[Json] | Json]:
        async with aclosing(self._source()) as source:
            async for event in source:
                if self.request.accepts(ContentType.event_stream):
                    yield SSEvent(event)
                else:
                    yield event

    async def _source(self) -> AsyncGenerator[Json]:
        yield {'a': 1}
        yield ['b', 2]

Run result

$ curl http://127.0.0.1:8000/api/user/events/ -X GET -H 'Accept: application/jsonl, application/json'
{"a":1}
["b",2]

$ curl http://127.0.0.1:8000/api/user/events/ -X GET -H 'Accept: text/event-stream, application/json'
data: {"a":1}

data: ["b",2]

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"
      },
      "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"
      },
      "SSEvent": {
        "description": "Server sent event payload.",
        "properties": {
          "comment": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "title": "Comment"
          },
          "data": {
            "title": "Data"
          },
          "event": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "title": "Event"
          },
          "id": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "string"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "title": "Id"
          },
          "retry": {
            "anyOf": [
              {
                "type": "integer"
              },
              {
                "type": "null"
              }
            ],
            "default": null,
            "title": "Retry"
          }
        },
        "required": [
          "data"
        ],
        "title": "SSEvent",
        "type": "object"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/eventstreaming/": {
      "get": {
        "operationId": "getEventStreamingApiEventstreaming",
        "responses": {
          "200": {
            "content": {
              "application/jsonl": {
                "itemSchema": {}
              },
              "text/event-stream": {
                "itemSchema": {
                  "$ref": "#/components/schemas/SSEvent"
                }
              }
            },
            "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 is a really advanced feature that is not required in 99% of cases. Here we define our own StreamingController subclass and override a special method to instantiate streaming renderers streaming_renderers().

Next we define an API endpoint for GET method. We use conditional_type() function to specify which type will be returned in which case for the OpenAPI metadata.

And _source() method which will provide events for both of the formats.

The last thing we do is we check what Accept header we are working with and provide an appropriate format for each case.

API Reference

Controllers

class dmr.streaming.controller.StreamingController(**kwargs)

Bases: Controller[_SerializerT_co]

Base class for all streaming controllers.

It can be used directly, but the most use-cases will be fine with just using the specific streaming protocol.

endpoint_cls

alias of _StreamingEndpoint

extras: ClassVar[Streaming] = Streaming(validate_events=EMPTY, ping_seconds=EMPTY)

Streaming settings for all endpoints of this controller.

Override it to change controller-level defaults, for example: extras = Streaming(validate_events=False) or extras = Streaming(ping_seconds=30). Per-endpoint values are passed as extras= to modify and validate. Use Streaming.of(self) to read the resolved values.

async handle_event_error(exc: Exception) → Any

Error handler for the events.

Is called when the StreamingResponse is iterated in the ASGI handler.

ping_event() → Any | None

Return a ping event to be generated if this streaming needs it.

By default pings are disabled for StreamingController types. Pings must be explicitly enabled in subclasses.

If ping_seconds is set in Streaming extras, this method will be called.

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

Returns the iterable of streaming renderers for this controller.

streaming_response_cls

Streaming response type to customize.

alias of StreamingResponse

to_stream(streaming_content: AsyncIterator[Any], *, status_code: HTTPStatus = HTTPStatus.OK, headers: Mapping[str, str] | None = None, cookies: Mapping[str, NewCookie] | None = None, regular_renderer: Renderer | None = None, streaming_renderer: StreamingRenderer | None = None, streaming_validator: StreamingValidator | None = None) → StreamingResponse

Convert streaming content to a streaming response.

Parameters:

• streaming_content – Async iterator which will produce the streaming events.

• status_code – Response status code. Defaults to 200.

• headers – Response extra headers.

• cookies – Response cookies.

• regular_renderer – Renderer instance to render event payloads.

• streaming_renderer – StreamingRenderer instance to render the event stream.

• streaming_validator – StreamingValidator instance to validate events one by one if validate_events is set to True.

Returns:

StreamingResponse instance with all the required properties set.

Endpoint

dmr.streaming.modify = ModifyEndpoint(extras_cls=<class 'dmr.streaming.endpoint.Streaming'>)

Decorator to modify endpoints that return raw model data.

Apply it to change some API parts:

>>> from http import HTTPStatus
>>> from dmr import Controller, modify
>>> from dmr.plugins.pydantic import PydanticFastSerializer

>>> class TaskController(Controller[PydanticFastSerializer]):
...     @modify(status_code=HTTPStatus.ACCEPTED)
...     def post(self) -> list[int]:
...         return [1, 2]  # id of tasks you have started

Parameters:

• status_code – Shows status_code in the documentation. When status_code is passed, always use it by default. When not provided, we use smart inference based on the HTTP method name for default returned response.

• headers – Shows headers in the documentation. When headers are passed we will add them for the default response.

• cookies – Shows cookies in the documentation. When cookies are passed we will add them for the default response.

• validate_responses – Do we have to run runtime validation of responses for this endpoint? Customizable via global setting, per controller, and per endpoint. Here we only store the per endpoint information.

• exclude_validate_responses – Set of status codes that we don’t validate, even when validate_responses is enabled. Useful for errors like 500 that can be raised from anywhere and that you might not want to describe. Overrides controller and settings values. Set it to None to validate all status codes back.

• semantic_schema – Should we generate any semantic schema for this endpoint?

• semantic_responses – Should semantic responses be collected from different providers for this endpoint.

• exclude_semantic_responses – Set of semantic responses status codes that user wants to disable. Overrides controller and settings values. Set it to None to enable all semantic responses back.

• semantic_auth – Should semantic auth be collected from different providers for this endpoint. Overrides controller and settings values.

• exclude_semantic_auth – Set of semantic security requirements names that should not be collected. Overrides controller and settings values.

• extra_responses – Sequence of extra responses that this endpoint can return. Overrides controller and settings values. Set it to None to only use the default response.

• no_validate_http_spec – Set of http spec validation checks that we disable for this endpoint. Overrides controller and settings values. Set it to None to enable all checks back.

• error_handler – Callback function to be called when this endpoint faces an exception.

• parsers – Sequence of types to be used for this endpoint to parse incoming request’s body. All types must be subtypes of Parser. Overrides controller and settings values.

• renderers – Sequence of types to be used for this endpoint to render response’s body. All types must be subtypes of Renderer. Overrides controller and settings values.

• validate_negotiation – Should we validate that returned response’s Content-Type header matches the one that we inferred in the negotiation process?

• auth – Sequence of auth instances to be used for this endpoint. Sync endpoints must use instances of dmr.security.SyncAuth. Async endpoints must use instances of dmr.security.AsyncAuth. Overrides controller and settings values. Set it to None to disable auth for this endpoint.

• throttling – Sequence of throttle instances to be used for this endpoint. Sync endpoints must use instances of dmr.throttling.SyncThrottle. Async endpoints must use instances of dmr.throttling.AsyncThrottle. Overrides controller and settings values. Set it to None to disable throttling of this endpoint.

• summary – A short summary of what the operation does. Defaults to the first paragraph of the endpoint’s docstring. Set it to None to have no summary at all.

• description – A verbose explanation of the operation behavior. Defaults to everything that goes after the first paragraph of the endpoint’s docstring. Set it to None to have no description at all.

• tags – A sequence of tags for API documentation control. Used to group operations in OpenAPI documentation. Overrides controller-level and router-level tags. Set it to None to have no tags at all.

• operation_id – Unique string used to identify the operation.

• deprecated – Declares this operation to be deprecated. Overrides controller-level and router-level values. Set it to False to explicitly mark this operation as not deprecated, even if its controller or router is.

• security – A sequence of security requirement objects for this operation. Overrides the controller value and security of dmr.openapi.OpenAPIConfig. Set it to None to disable security of this endpoint. It is merged with the requirements generated from auth, but it never affects the runtime auth. Useful to document external security mechanisms, for example, the ones enforced by an HTTP proxy. Used security schemes must be declared as security_schemes in components of dmr.openapi.OpenAPIConfig. See Customizing security.

• external_docs – Additional external documentation for this operation. Overrides the controller value. Set it to None to have no external docs at all.

• callbacks – A map of possible out-of band callbacks related to the parent operation. The key is a unique identifier for the Callback Object. Each value in the map is a Callback Object that describes a request that may be initiated by the API provider and the expected responses. Overrides the controller value. Set it to None to have no callbacks at all.

• servers – An alternative servers sequence to service this operation. Overrides the controller value. Set it to None to have no servers at all.

• x_extensions – Specification extensions of this operation, keys must start with x-. Set only on this operation: it is never inherited from or merged with x_extensions, which describes the path item, not its operations.

• links – Possible links to other OpenAPI operations.

• response_description – Description for the generated response object.

• ignore_from_spec – If set to True, this endpoint would not be added to the final OpenAPI spec.

• extras – Extra settings for custom controllers. See Providing extras for @modify and @validate to learn more.

Returns:

The same function with __dmr_payload__ payload instance.

Warning

Do not disable validate_responses unless this is performance critical for you!

Changed in version 0.15.0: modify used to be a function, now it is an instance with lazy method for lazy reusable endpoints.

Changed in version 0.16.0: Removed validate_events parameter, added extras parameter instead. This class is now generic and public.

dmr.streaming.validate = ValidateEndpoint(extras_cls=<class 'dmr.streaming.endpoint.Streaming'>)

Decorator to validate responses from endpoints that return HttpResponse.

Apply it to validate important API parts:

>>> from http import HTTPStatus
>>> from django.http import HttpResponse
>>> from dmr import Controller, validate, ResponseSpec
>>> from dmr.plugins.pydantic import PydanticFastSerializer

>>> class TaskController(Controller[PydanticFastSerializer]):
...     @validate(
...         ResponseSpec(
...             return_type=list[int],
...             status_code=HTTPStatus.OK,
...         ),
...     )
...     def post(self) -> HttpResponse:
...         return HttpResponse(b'[1, 2]', status=HTTPStatus.OK)

Response validation can be disabled for extra speed by sending validate_responses falsy parameter or by setting this configuration in your settings.py file:

settings.py

>>> DMR_SETTINGS = {'validate_responses': False}

Parameters:

• response – The main response that this endpoint is allowed to return.

• responses – A collection of other responses that are allowed to be returned from this endpoint.

• validate_responses – Do we have to run runtime validation of responses for this endpoint? Customizable via global setting, per controller, and per endpoint. Here we only store the per endpoint information.

• exclude_validate_responses – Set of status codes that we don’t validate, even when validate_responses is enabled. Useful for errors like 500 that can be raised from anywhere and that you might not want to describe. Overrides controller and settings values. Set it to None to validate all status codes back.

• semantic_schema – Should we generate any semantic schema for this endpoint?

• semantic_responses – Should semantic responses be collected from different providers for this endpoint. Overrides controller and settings values.

• exclude_semantic_responses – Set of semantic responses status codes that user wants to disable. Overrides controller and settings values. Set it to None to enable all semantic responses back.

• semantic_auth – Should semantic auth be collected from different providers for this endpoint. Overrides controller and settings values.

• exclude_semantic_auth – Set of semantic security requirements names that should not be collected. Overrides controller and settings values.

• no_validate_http_spec – Set of http spec validation checks that we disable for this endpoint. Overrides controller and settings values. Set it to None to enable all checks back.

• error_handler – Callback function to be called when this endpoint faces an exception.

• parsers – Sequence of types to be used for this endpoint to parse incoming request’s body. All types must be subtypes of Parser. Overrides controller and settings values.

• renderers – Sequence of types to be used for this endpoint to render response’s body. All types must be subtypes of Renderer. Overrides controller and settings values.

• validate_negotiation – Should we validate that returned response’s Content-Type header matches the one that we inferred in the negotiation process?

• auth – Sequence of auth instances to be used for this endpoint. Sync endpoints must use instances of dmr.security.SyncAuth. Async endpoints must use instances of dmr.security.AsyncAuth. Overrides controller and settings values. Set it to None to disable auth for this endpoint.

• throttling – Sequence of throttle instances to be used for this endpoint. Sync endpoints must use instances of dmr.throttling.SyncThrottle. Async endpoints must use instances of dmr.throttling.AsyncThrottle. Overrides controller and settings values. Set it to None to disable throttling of this endpoint.

• summary – A short summary of what the operation does. Defaults to the first paragraph of the endpoint’s docstring. Set it to None to have no summary at all.

• description – A verbose explanation of the operation behavior. Defaults to everything that goes after the first paragraph of the endpoint’s docstring. Set it to None to have no description at all.

• tags – A sequence of tags for API documentation control. Used to group operations in OpenAPI documentation. Overrides controller-level and router-level tags. Set it to None to have no tags at all.

• operation_id – Unique string used to identify the operation.

• deprecated – Declares this operation to be deprecated. Overrides controller-level and router-level values. Set it to False to explicitly mark this operation as not deprecated, even if its controller or router is.

• security – A sequence of security requirement objects for this operation. Overrides the controller value and security of dmr.openapi.OpenAPIConfig. Set it to None to disable security of this endpoint. It is merged with the requirements generated from auth, but it never affects the runtime auth. Useful to document external security mechanisms, for example, the ones enforced by an HTTP proxy. Used security schemes must be declared as security_schemes in components of dmr.openapi.OpenAPIConfig. See Customizing security.

• external_docs – Additional external documentation for this operation. Overrides the controller value. Set it to None to have no external docs at all.

• callbacks – A map of possible out-of band callbacks related to the parent operation. The key is a unique identifier for the Callback Object. Each value in the map is a Callback Object that describes a request that may be initiated by the API provider and the expected responses. Overrides the controller value. Set it to None to have no callbacks at all.

• servers – An alternative servers sequence to service this operation. Overrides the controller value. Set it to None to have no servers at all.

• x_extensions – Specification extensions of this operation, keys must start with x-. Set only on this operation: it is never inherited from or merged with x_extensions, which describes the path item, not its operations.

• ignore_from_spec – If set to True, this endpoint would not be added to the final OpenAPI spec.

• extras – Extra settings for custom controllers. See Providing extras for @modify and @validate to learn more.

Returns:

The same function with __dmr_payload__ payload instance.

Warning

Do not disable validate_responses unless this is performance critical for you!

Changed in version 0.15.0: validate used to be a function, now it is an instance with lazy method for lazy reusable endpoints.

Changed in version 0.16.0: Removed validate_events parameter, added extras parameter instead. This class is now generic and public.

final class dmr.streaming.Streaming(*, validate_events: bool | EMPTY = EMPTY, ping_seconds: float | EMPTY | None = EMPTY)

Extra settings for endpoints of streaming controllers.

Pass it as extras= to modify or validate. It can only be used with streaming controllers.

Set it as extras on a controller to provide controller-level defaults, for example: extras = Streaming(validate_events=False). A controller that assigns its own extras replaces all values of its base controllers, the same way any other attribute does.

validate_events

Should this endpoint validate events? If not set, defaults to the controller value, then to validate_events, then to the validate_responses value.

Type:

bool | EMPTY

ping_seconds

Optional ping keep alive event support. Some servers might close long living connections with no activity. Specify number of seconds to wait for the next event before sending a ping event. The payload of the ping event is defined in ping_event(). None disables pings. If not set, defaults to the controller value, then to None. SSEController enables pings every 15 seconds by default.

Type:

float | EMPTY | None

Added in version 0.16.0.

classmethod build(from_endpoint: Self | EMPTY, from_controller: Self, controller_cls: type[Controller[BaseSerializer]], builder: EndpointMetadataBuilder) → StreamingExtras

Resolve streaming settings from all configuration layers.

final class dmr.streaming.endpoint.StreamingExtras(*, validate_events: bool, ping_seconds: float | None)

Resolved streaming settings of an endpoint.

It is stored in extras for all endpoints of streaming controllers.

validate_events

Should this endpoint validate events?

Type:

bool

ping_seconds

Seconds to wait for the next event before sending a ping event, None disables pings.

Type:

float | None

Added in version 0.16.0.

Responses

class dmr.streaming.stream.StreamingResponse(streaming_content: AsyncIterator[Any], controller: StreamingController[BaseSerializer], *, regular_renderer: Renderer, streaming_renderer: StreamingRenderer, streaming_validator: StreamingValidator, headers: Mapping[str, str] | None = None, status_code: HTTPStatus = HTTPStatus.OK)

Bases: HttpResponseBase

Our own response subclass to mark that we explicitly return SSE.

Converts events to bytes with the help of a passed serializer and renderer types.

We don’t inherit from django.http.StreamingHttpResponse here, because it has a strict API for streaming that we can’t use or customize.

__aiter__() → AsyncIterator[bytes]

ASGI handler protocol for streaming responses.

When streaming is True, ASGI handler will async iterate over the response object.

When doing so, we will be inside the ASGI handler already. No DMR error handling will work.

__iter__() → Iterator[bytes]

In development it is useful to have sync interface for streaming.

This is a part of the WSGI handler protocol.

Danger

Do not use this in production! We even added a special error to catch this. In production you must use ASGI servers like uvicorn with streaming.

This implementation has a lot of limitations. Be careful even in development.

close() → None

Closes the response and cleans up all references.

streaming: Final = True

Part of the ASGI handler protocol. Will trigger __aiter__

Renderers

class dmr.streaming.renderer.StreamingRenderer(serializer: type[BaseSerializer], regular_renderer: Renderer, streaming_validator_cls: type[StreamingValidator])

Bases: Renderer

Base class for all streaming responses.

It is different from the regular Renderer in several ways:

1. We need to initialize this renderer with a subrenderer, which will render the individual events itself

2. Serializer is needed to serialize events

3. Validator is needed to optionally validate events

streaming: ClassVar[Literal[True]] = True

Whether or not this renderer is used for streaming responses.

property validation_parser: _NoOpParser

Returns a parser that can parse what this renderer rendered.

Why? Because when validate_responses is True, we parse the response body once again to see if it fits the schema.

That’s why all renderers must know how to unparse its results.

Validation

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

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

This is very different from any other validator. Why?

1. Because we send just one response. No events can be produced at all for a long period of time. Some events can be correct, while other can be wrong

2. We can’t close the connection when finding wrong events, it will be a big problem for our users and it would be hard to debug

3. But, we can modify events to be error events instead!

4. When validation is active and the event is either not the model we expect or has the wrong payload type - we send the error event

apply_event_pipeline(event: Any) → Any

Runs the pipeline.

classmethod from_controller(controller: StreamingController[BaseSerializer], status_code: HTTPStatus) → Self

Construct validator from a controller instance.

Inferences event type model from the endpoint metadata. Also knows whether or not the events validation is turned on or not.

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

Abstract method to define the event validation pipeline.

class dmr.streaming.validation.StreamingResponseValidator(metadata: EndpointMetadata, serializer: type[BaseSerializer])

Streaming response validator.

Added in version 0.16.0.

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

Validate that the event type matches the model.