orionis.http.adapters.response.asgi.ASGIResponseAdapter

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

class documentation

class ASGIResponseAdapter(ResponseAdapter): (source)

View In Hierarchy

Undocumented

Method	`__ensureContentLength`	Add content-length to headers if absent, reflecting the body size.
Async Method	`__fileRangeIterator`	Yield file bytes within [start, end) range asynchronously.
Async Method	`__sendHead`	Send HEAD framing without starting an event producer.
Async Method	`__sendStream`	Send and close a stream with transport backpressure.
Async Method	`__waitDisconnect`	Wait for disconnect, draining unconsumed request messages.
Async Method	`send`	Send the HTTP response using the ASGI protocol.
Constant	`RESPONSE_BODY`	Undocumented
Constant	`RESPONSE_START`	Undocumented
Class Variable	`__slots__`	Undocumented

def __ensureContentLength(self, headers: list[tuple[bytes, bytes]], response: Response): (source) ¶

Add content-length to headers if absent, reflecting the body size.

Parameters
headers:`list of tuple of bytes`	Mutable headers list to append content-length into.
response:`Response`	Response object used to compute the expected body size.

Returns
`None`	Headers list is mutated in place; no value is returned.

async def __fileRangeIterator(self, path: Path, start: int, end: int, chunk_size: int = 64 * 1024) -> AsyncGenerator[bytes]: (source) ¶

Yield file bytes within [start, end) range asynchronously.

Parameters
path:`Path`	Path to the file to read.
start:`int`	Byte offset to start reading from.
end:`int`	Exclusive byte offset to stop reading at.
chunk_size:`int, default 65536`	Number of bytes to read per chunk.

Returns
`AsyncGenerator[bytes]`	Asynchronous generator yielding file chunks.

async def __sendHead(self, response: Response, status: int, headers: list[tuple[bytes, bytes]], send: Callable[..., Awaitable[None]]): (source) ¶

Send HEAD framing without starting an event producer.

Parameters
response:`Response`	Response whose metadata is sent.
status:`int`	Response status code.
headers:`list[tuple[bytes, bytes]]`	Encoded response headers.
send:`Callable[..., Awaitable[None]]`	ASGI response writer.

Returns
`None`	Close an unstarted event stream, then run successful background work.

async def __sendStream(self, iterator: AsyncIterator[bytes], status: int, headers: list[tuple[bytes, bytes]], send: Callable[..., Awaitable[None]]): (source) ¶

Send and close a stream with transport backpressure.

Parameters
iterator:`AsyncIterator[bytes]`	Owned iterator to close even if sending headers fails.
status:`int`	Response status code.
headers:`list[tuple[bytes, bytes]]`	Encoded response headers.
send:`Callable[..., Awaitable[None]]`	ASGI response writer.

Returns
`None`	Finish the body only after iteration and cleanup succeed.

async def __waitDisconnect(self, receive: Callable[..., Awaitable[dict]]): (source) ¶

Wait for disconnect, draining unconsumed request messages.

Parameters
receive:`Callable[..., Awaitable[dict]]`	ASGI receive channel, owned by the response after handler return.

Returns
`None`	Return only when the client disconnects.

Raises
`RuntimeError`	If the HTTP channel produces an unexpected message type.

async def send(self, adapter: TransportAdapter, response: Response, _receive: Callable[..., Awaitable[dict]], send: Callable[..., Awaitable[None]]): (source) ¶

overrides orionis.http.adapters.response.contracts.response.ResponseAdapter.send

Send the HTTP response using the ASGI protocol.

Parameters
adapter:`TransportAdapter`	Transport adapter containing request information.
response:`Response`	Response object to be sent back to the client.
_receive:`Callable[..., Awaitable[dict]]`	Receive callable observed for disconnects during event streams. Handlers must consume required request body data before returning an event stream, which owns receive during response delivery.
send:`Callable[..., Awaitable[None]]`	Awaitable callable to send ASGI messages.

Returns
`None`	Sends the response via ASGI protocol and returns nothing.

Raises
`BaseException`	Propagate metadata or delivery failures after closing an owned SSE source.

RESPONSE_BODY: str = (source) ¶

Undocumented

Value

'http.response.body'

RESPONSE_START: str = (source) ¶

Undocumented

Value

'http.response.start'

__slots__: tuple = (source) ¶

overrides orionis.http.adapters.response.contracts.response.ResponseAdapter.__slots__

Undocumented