def __delattr__(self, key: Any) -> None:
del self._state[key]
+
+ def __getitem__(self, key: str) -> Any:
+ return self._state[key]
+
+ def __setitem__(self, key: str, value: Any) -> None:
+ self._state[key] = value
+
+ def __delitem__(self, key: str) -> None:
+ del self._state[key]
+
+ def __iter__(self) -> Iterator[str]:
+ return iter(self._state)
+
+ def __len__(self) -> int:
+ return len(self._state)
import json
from collections.abc import AsyncGenerator, Iterator, Mapping
from http import cookies as http_cookies
-from typing import TYPE_CHECKING, Any, NoReturn, cast
+from typing import TYPE_CHECKING, Any, Generic, NoReturn, cast
import anyio
+from typing_extensions import TypeVar
from starlette._utils import AwaitableOrContextManager, AwaitableOrContextManagerWrapper
from starlette.datastructures import URL, Address, FormData, Headers, QueryParams, State
pass
-class HTTPConnection(Mapping[str, Any]):
+StateT = TypeVar("StateT", bound=Mapping[str, Any] | State, default=State)
+
+
+class HTTPConnection(Mapping[str, Any], Generic[StateT]):
"""
A base class for incoming HTTP connections, that is used to provide
any functionality that is common to both `Request` and `WebSocket`.
return self.scope["user"]
@property
- def state(self) -> State:
+ def state(self) -> StateT:
if not hasattr(self, "_state"):
# Ensure 'state' has an empty dict if it's not already populated.
self.scope.setdefault("state", {})
# Create a state instance with a reference to the dict in which it should
# store info
self._state = State(self.scope["state"])
- return self._state
+ return cast(StateT, self._state)
def url_for(self, name: str, /, **path_params: Any) -> URL:
url_path_provider: Router | Starlette | None = self.scope.get("router") or self.scope.get("app")
raise RuntimeError("Send channel has not been made available")
-class Request(HTTPConnection):
+class Request(HTTPConnection[StateT]):
_form: FormData | None
def __init__(self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send):