connection#

Some code in this module was adapted from encode/starlette and encode/starlette.

Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:

Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS “AS IS” AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

class litestar.connection.ASGIConnection#

Bases: Generic[HandlerT, UserT, AuthT, StateT]

The base ASGI connection container.

__init__(scope: Scope, receive: Receive = <function empty_receive>, send: Send = <function empty_send>) → None#

Initialize ASGIConnection.

Parameters:

scope¶ – The ASGI connection scope.
receive¶ – The ASGI receive function.
send¶ – The ASGI send function.

scope: Scope#: The ASGI scope attached to the connection.

receive: Receive#: The ASGI receive function.

send: Send#: The ASGI send function.

property app: Litestar#

Return the app for this connection.

Returns:: The Litestar application instance

property route_handler: HandlerT#

Return the route_handler for this connection.

Returns:: The target route handler instance.

property state: StateT#

Return the State of this connection.

Returns:: A State instance constructed from the scope[“state”] value.

property url: URL#

Return the URL of this connection’s Scope.

Returns:: A URL instance constructed from the request’s scope.

property base_url: URL#

Return the base URL of this connection’s Scope.

Returns:: A URL instance constructed from the request’s scope, representing only the base part (host + domain + prefix) of the request.

property headers: Headers#

Return the headers of this connection’s Scope.

Returns:: A Headers instance with the request’s scope[“headers”] value.

property query_params: MultiDict[Any]#

Return the query parameters of this connection’s Scope.

Returns:: A normalized dict of query parameters. Multiple values for the same key are returned as a list.

property path_params: dict[str, Any]#

Return the path_params of this connection’s Scope.

Returns:: A string keyed dictionary of path parameter values.

property cookies: dict[str, str]#

Return the cookies of this connection’s Scope.

Returns:: Returns any cookies stored in the header as a parsed dictionary.

property client: Address | None#

Return the client data of this connection’s Scope.

Returns:: A two tuple of the host name and port number.

property auth: AuthT#

Return the auth data of this connection’s Scope.

Raises:: ImproperlyConfiguredException – If auth is not set in scope via an AuthMiddleware, raises an exception
Returns:: A type correlating to the generic variable Auth.

property user: UserT#

Return the user data of this connection’s Scope.

Raises:: ImproperlyConfiguredException – If user is not set in scope via an AuthMiddleware, raises an exception
Returns:: A type correlating to the generic variable User.

property session: dict[str, Any]#

Return the session for this connection if a session was previously set in the Scope

Returns:: A dictionary representing the session value - if existing.
Raises:: ImproperlyConfiguredException – if session is not set in scope.

property logger: Logger#

Return the Logger instance for this connection.

Returns:: A Logger instance.
Raises:: ImproperlyConfiguredException – if log_config has not been passed to the Litestar constructor.

set_session(value: dict[str, Any] | DataContainerType | EmptyType) → None#

Set the session in the connection’s Scope.

If the SessionMiddleware is enabled, the session will be added to the response as a cookie header.

Parameters:: value¶ – Dictionary or pydantic model instance for the session data.
Returns:: None

clear_session() → None#

Remove the session from the connection’s Scope.

If the Litestar SessionMiddleware is enabled, this will cause the session data to be cleared.

Returns:: None.

url_for(name: str, **path_parameters: Any) → str#

Return the url for a given route handler name.

Parameters:

name¶ – The name of the request route handler.
**path_parameters¶ – Values for path parameters in the route

Raises:

NoRouteMatchFoundException – If route with name does not exist, path parameters are missing or have a wrong type.

Returns:

A string representing the absolute url of the route handler.

url_for_static_asset(name: str, file_path: str) → str#

Receives a static files handler name, an asset file path and returns resolved absolute url to the asset.

Parameters:

name¶ – A static handler unique name.
file_path¶ – a string containing path to an asset.

Raises:

NoRouteMatchFoundException – If static files handler with name does not exist.

Returns:

A string representing absolute url to the asset.

class litestar.connection.Request#

Bases: Generic[UserT, AuthT, StateT], ASGIConnection[HTTPRouteHandler, UserT, AuthT, StateT]

The Litestar Request class.

__init__(scope: Scope, receive: Receive = <function empty_receive>, send: Send = <function empty_send>) → None#

Initialize Request.

Parameters:

scope¶ – The ASGI connection scope.
receive¶ – The ASGI receive function.
send¶ – The ASGI send function.

property method: Method#

Return the request method.

Returns:: The request Method

property content_type: tuple[str, dict[str, str]]#

Parse the request’s ‘Content-Type’ header, returning the header value and any options as a dictionary.

Returns:: A tuple with the parsed value and a dictionary containing any options send in it.

property accept: Accept#

Parse the request’s ‘Accept’ header, returning an Accept instance.

Returns:: An Accept instance, representing the list of acceptable media types.

async json() → Any#

Retrieve the json request body from the request.

Returns:: An arbitrary value

async msgpack() → Any#

Retrieve the MessagePack request body from the request.

Returns:: An arbitrary value

async stream() → AsyncGenerator[bytes, None]#

Return an async generator that streams chunks of bytes.

Returns:: An async generator.
Raises:: RuntimeError – if the stream is already consumed

async body() → bytes#

Return the body of the request.

Returns:: A byte-string representing the body of the request.

async form() → FormMultiDict#

Retrieve form data from the request. If the request is either a ‘multipart/form-data’ or an ‘application/x-www-form- urlencoded’, return a FormMultiDict instance populated with the values sent in the request, otherwise, an empty instance.

Returns:: A FormMultiDict instance

async send_push_promise(path: str, raise_if_unavailable: bool = False) → None#

Send a push promise.

This method requires the http.response.push extension to be sent from the ASGI server.

Parameters:

path¶ – Path to send the promise to.
raise_if_unavailable¶ – Raise an exception if server push is not supported by the server

Returns:

None

class litestar.connection.WebSocket#

Bases: Generic[UserT, AuthT, StateT], ASGIConnection[WebsocketRouteHandler, UserT, AuthT, StateT]

The Litestar WebSocket class.

__init__(scope: Scope, receive: Receive = <function empty_receive>, send: Send = <function empty_send>) → None#

Initialize WebSocket.

Parameters:

scope¶ – The ASGI connection scope.
receive¶ – The ASGI receive function.
send¶ – The ASGI send function.

receive_wrapper(receive: Receive) → Receive#

Wrap receive to set ‘self.connection_state’ and validate events.

Parameters:: receive¶ – The ASGI receive function.
Returns:: An ASGI receive function.

send_wrapper(send: Send) → Send#

Wrap send to ensure that state is not disconnected.

Parameters:: send¶ – The ASGI send function.
Returns:: An ASGI send function.

async accept(subprotocols: str | None = None, headers: Headers | dict[str, Any] | list[tuple[bytes, bytes]] | None = None) → None#

Accept the incoming connection. This method should be called before receiving data.

Parameters:

subprotocols¶ – Websocket sub-protocol to use.
headers¶ – Headers to set on the data sent.

Returns:

None

async close(code: int = 1000, reason: str | None = None) → None#

Send an ‘websocket.close’ event.

Parameters:

code¶ – Status code.
reason¶ – Reason for closing the connection

Returns:

None

async receive_data(mode: Literal['text']) → str#

async receive_data(mode: Literal['binary']) → bytes

Receive an ‘websocket.receive’ event and returns the data stored on it.

Parameters:: mode¶ – The respective event key to use.
Returns:: The event’s data.

async iter_data(mode: Literal['text']) → AsyncGenerator[str, None]#

async iter_data(mode: Literal['binary']) → AsyncGenerator[bytes, None]

Continuously receive data and yield it

Parameters:: mode¶ – Socket mode to use. Either text or binary

async receive_text() → str#

Receive data as text.

Returns:: A string.

async receive_bytes() → bytes#

Receive data as bytes.

Returns:: A byte-string.

async receive_json(mode: WebSocketMode = 'text') → Any#

Receive data and decode it as JSON.

Parameters:: mode¶ – Either text or binary.
Returns:: An arbitrary value

async receive_msgpack() → Any#

Receive data and decode it as MessagePack.

Note that since MessagePack is a binary format, this method will always receive data in binary mode.

Returns:: An arbitrary value

async iter_json(mode: WebSocketMode = 'text') → AsyncGenerator[Any, None]#

Continuously receive data and yield it, decoding it as JSON in the process.

Parameters:: mode¶ – Socket mode to use. Either text or binary

async iter_msgpack() → AsyncGenerator[Any, None]#

Continuously receive data and yield it, decoding it as MessagePack in the process.

Note that since MessagePack is a binary format, this method will always receive data in binary mode.

async send_data(data: str | bytes, mode: WebSocketMode = 'text', encoding: str = 'utf-8') → None#

Send a ‘websocket.send’ event.

Parameters:

data¶ – Data to send.
mode¶ – The respective event key to use.
encoding¶ – Encoding to use when converting bytes / str.

Returns:

None

async send_text(data: bytes, encoding: str = 'utf-8') → None#

async send_text(data: str) → None

Send data using the text key of the send event.

Parameters:

data¶ – Data to send
encoding¶ – Encoding to use for binary data.

Returns:

None

async send_bytes(data: bytes) → None#

async send_bytes(data: str, encoding: str = 'utf-8') → None

Send data using the bytes key of the send event.

Parameters:

data¶ – Data to send
encoding¶ – Encoding to use for binary data.

Returns:

None

async send_json(data: Any, mode: WebSocketMode = 'text', encoding: str = 'utf-8', serializer: Serializer = <function default_serializer>) → None#

Send data as JSON.

Parameters:

data¶ – A value to serialize.
mode¶ – Either text or binary.
encoding¶ – Encoding to use for binary data.
serializer¶ – A serializer function.

Returns:

None

async send_msgpack(data: Any, encoding: str = 'utf-8', serializer: Serializer = <function default_serializer>) → None#

Send data as MessagePack.

Note that since MessagePack is a binary format, this method will always send data in binary mode.

Parameters:

data¶ – A value to serialize.
encoding¶ – Encoding to use for binary data.
serializer¶ – A serializer function.

Returns:

None