Stream class

esmerald.datastructures.stream.Stream

Bases: ResponseContainer[StreamingResponse]

iterator instance-attribute

iterator

Any iterable function.

to_response

to_response(headers, media_type, status_code, app)

PARAMETER	DESCRIPTION
headers	TYPE: Dict[str, Any]
media_type	TYPE: Union[MediaType, str]
status_code	TYPE: int
app	TYPE: Type[Esmerald]

Source code in esmerald/datastructures/stream.py

def to_response(
    self,
    headers: Dict[str, Any],
    media_type: Union["MediaType", str],
    status_code: int,
    app: Type["Esmerald"],
) -> StreamingResponse:  # pragma: no cover
    return StreamingResponse(
        background=self.background,
        content=(
            self.iterator
            if isinstance(self.iterator, (Iterable, AsyncIterable))
            else self.iterator()
        ),
        headers=headers,
        media_type=media_type,
        status_code=status_code,
    )