ORIONIS API REFERENCE

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

orionis.http.EventStreamResponse

class documentation

class EventStreamResponse(StreamingResponse): (source)

Constructor: EventStreamResponse(content, status_code, headers, background)

View In Hierarchy

Stream typed server-sent events through the existing response transport.

Method __init__ Initialize a lazy UTF-8 event stream with SSE response headers.
Method getRawHeaders Serialize headers without a fixed event-stream content length.
Method getStream Return the owned byte iterator without starting its event source.
Method getStringHeaders Serialize headers without a fixed event-stream content length.
Class Variable __slots__ Undocumented

Inherited from StreamingResponse:

Async Method _wrapSyncIterable Adapt a synchronous iterable of bytes to an asynchronous iterable.
Instance Variable _body Undocumented

Inherited from Response (via StreamingResponse):

Method addHeader Add a header to the response.
Method deleteCookie Delete a cookie by setting its expiration to the past.
Method getBody Return the response body as bytes.
Method getFlashData Return the data queued by withFlash() for the session flash bag.
Method getHeader Get the values for a header.
Method getMediaType Return the media type of the response.
Method getStatusCode Return the HTTP status code of the response.
Method hasHeader Check if a header is present.
Method hasStream Check if the response has a stream.
Method removeHeader Remove a header from the response.
Method render Render the content to bytes.
Async Method runBackground Run the background task if it exists.
Method setCookie Set a cookie header in the response.
Method setHeader Set a header, replacing any existing values.
Method withCookie Attach a cookie and return the response for chaining.
Method withCookies Attach several cookies at once and return the response.
Method withErrors Flash validation errors for the next request.
Method withFlash Flash a status message into the session and return the response.
Method withInput Flash the submitted payload so the next request can repopulate it.
Method withoutCookie Expire a cookie and return the response for chaining.
Class Variable charset Undocumented
Instance Variable _flash Undocumented
Instance Variable _headers Undocumented
Instance Variable _raw_headers Undocumented
Instance Variable _stream Undocumented
Instance Variable background Undocumented
Instance Variable media_type Undocumented
Instance Variable status_code Undocumented
def __init__(self, content: AsyncIterable[ServerSentEvent | str] | Iterable[ServerSentEvent | str], status_code: HTTPStatus | int = 200, headers: Mapping[str, str] | None = None, background: BackgroundTask | None = None): (source)
overrides orionis.http.StreamingResponse.__init__

Initialize a lazy UTF-8 event stream with SSE response headers.

Parameters
content:AsyncIterable[ServerSentEvent | str] | Iterable[ServerSentEvent | str]Events or strings, each encoded only when the transport requests it.
status_code:HTTPStatus | int, optionalHTTP status code for the response.
headers:Mapping[str, str] | None, optionalResponse headers; Content-Length is discarded for event streams.
background:BackgroundTask | None, optionalTask to run after successful delivery and producer cleanup.
Returns
NoneValidate response metadata while deferring producer iteration and I/O.
Raises
TypeErrorIf the content does not implement synchronous or asynchronous iteration.
def getRawHeaders(self) -> list[tuple[bytes, bytes]]: (source)
overrides orionis.http.Response.getRawHeaders

Serialize headers without a fixed event-stream content length.

Returns
list of tuple of bytesASGI response headers, excluding any later Content-Length mutation.
def getStream(self) -> _EventStreamIterator: (source)
overrides orionis.http.Response.getStream

Return the owned byte iterator without starting its event source.

Returns
_EventStreamIteratorLazy encoder whose aclose() method finalizes the event source.
def getStringHeaders(self) -> list[tuple[str, str]]: (source)
overrides orionis.http.Response.getStringHeaders

Serialize headers without a fixed event-stream content length.

Returns
list of tuple of strRSGI response headers, excluding any later Content-Length mutation.
__slots__: tuple = (source)
overrides orionis.http.StreamingResponse.__slots__

Undocumented