Remove invalid uses of AnyStr

src/quart/app.py

@@ -15,7 +15,6 @@
 from inspect import isgenerator
 from types import TracebackType
 from typing import Any
-from typing import AnyStr
 from typing import Callable
 from typing import cast
 from typing import NoReturn
@@ -1304,7 +1303,7 @@ def test_request_context(
         query_string: dict | None = None,
         scheme: str = "http",
         send_push_promise: Callable[[str, Headers], Awaitable[None]] = no_op_push,
-        data: AnyStr | None = None,
+        data: str | bytes | None = None,
         form: dict | None = None,
         json: Any = sentinel,
         root_path: str = "",

src/quart/asgi.py

@@ -3,7 +3,6 @@
 import asyncio
 import warnings
 from functools import partial
-from typing import AnyStr
 from typing import cast
 from typing import Optional
 from typing import TYPE_CHECKING
@@ -306,7 +305,7 @@ async def handle_websocket(
                 cast(WebsocketCloseEvent, {"type": "websocket.close", "code": 1000})
             )
 
-    async def send_data(self, send: ASGISendCallable, data: AnyStr) -> None:
+    async def send_data(self, send: ASGISendCallable, data: str | bytes) -> None:
         if isinstance(data, str):
             await send({"type": "websocket.send", "bytes": None, "text": data})
         else:

src/quart/signals.py

@@ -44,11 +44,11 @@
 websocket_started = _signals.signal("websocket-started")
 
 #: Called on receipt of a message over the websocket, connected
-# functions should have a signature of Callable[[AnyStr], None]
+# functions should have a signature of Callable[[str | bytes], None]
 websocket_received = _signals.signal("websocket-received")
 
 #: Called when a message has been sent over the websocket, connected
-# functions should have a signature of Callable[[AnyStr], None]
+# functions should have a signature of Callable[[str | bytes], None]
 websocket_sent = _signals.signal("websocket-sent")
 
 #: Called after a response is fully finalised, connected functions

src/quart/testing/client.py

@@ -7,7 +7,6 @@
 from http.cookiejar import CookieJar
 from types import TracebackType
 from typing import Any
-from typing import AnyStr
 from typing import TYPE_CHECKING
 from urllib.request import Request as U2Request
 
@@ -76,7 +75,7 @@ async def open(
         *,
         method: str = "GET",
         headers: dict | Headers | None = None,
-        data: AnyStr | None = None,
+        data: str | bytes | None = None,
         form: dict | None = None,
         files: dict[str, FileStorage] | None = None,
         query_string: dict | None = None,
@@ -328,7 +327,7 @@ async def session_transaction(
         headers: dict | Headers | None = None,
         query_string: dict | None = None,
         scheme: str = "http",
-        data: AnyStr | None = None,
+        data: str | bytes | None = None,
         form: dict | None = None,
         json: Any = sentinel,
         root_path: str = "",
@@ -406,7 +405,7 @@ async def _make_request(
         path: str,
         method: str,
         headers: dict | Headers | None,
-        data: AnyStr | None,
+        data: str | bytes | None,
         form: dict | None,
         files: dict[str, FileStorage] | None,
         query_string: dict | None,

src/quart/testing/connections.py

@@ -4,7 +4,6 @@
 from collections.abc import Awaitable
 from types import TracebackType
 from typing import Any
-from typing import AnyStr
 from typing import TYPE_CHECKING
 
 from hypercorn.typing import ASGIReceiveEvent
@@ -146,14 +145,14 @@ async def __aexit__(
             ):
                 raise data
 
-    async def receive(self) -> AnyStr:
+    async def receive(self) -> str | bytes:
         data = await self._receive_queue.get()
         if isinstance(data, Exception):
             raise data
         else:
             return data
 
-    async def send(self, data: AnyStr) -> None:
+    async def send(self, data: str | bytes) -> None:
         if isinstance(data, str):
             await self._send_queue.put({"type": "websocket.receive", "text": data})
         else:

src/quart/testing/utils.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from typing import Any
-from typing import AnyStr
 from typing import cast
 from typing import Literal
 from typing import overload
@@ -80,7 +79,7 @@ def make_test_headers_path_and_query_string(
 
 def make_test_body_with_headers(
     *,
-    data: AnyStr | None = None,
+    data: str | bytes | None = None,
     form: dict | None = None,
     files: dict[str, FileStorage] | None = None,
     json: Any = sentinel,

src/quart/typing.py

@@ -13,7 +13,6 @@
 from http.cookiejar import CookieJar
 from types import TracebackType
 from typing import Any
-from typing import AnyStr
 from typing import Callable
 from typing import Optional
 from typing import TYPE_CHECKING
@@ -182,9 +181,9 @@ async def __aexit__(
         self, exc_type: type, exc_value: BaseException, tb: TracebackType
     ) -> None: ...
 
-    async def receive(self) -> AnyStr: ...
+    async def receive(self) -> str | bytes: ...
 
-    async def send(self, data: AnyStr) -> None: ...
+    async def send(self, data: str | bytes) -> None: ...
 
     async def receive_json(self) -> Any: ...
 
@@ -210,7 +209,7 @@ async def open(
         *,
         method: str = "GET",
         headers: dict | Headers | None = None,
-        data: AnyStr | None = None,
+        data: str | bytes | None = None,
         form: dict | None = None,
         files: dict[str, FileStorage] | None = None,
         query_string: dict | None = None,
@@ -297,7 +296,7 @@ def session_transaction(
         headers: dict | Headers | None = None,
         query_string: dict | None = None,
         scheme: str = "http",
-        data: AnyStr | None = None,
+        data: str | bytes | None = None,
         form: dict | None = None,
         json: Any = None,
         root_path: str = "",

src/quart/wrappers/websocket.py

@@ -2,7 +2,6 @@
 
 import asyncio
 from typing import Any
-from typing import AnyStr
 from typing import Callable
 
 from hypercorn.typing import WebsocketScope
@@ -55,11 +54,11 @@ def __init__(
     def requested_subprotocols(self) -> list[str]:
         return self._subprotocols
 
-    async def receive(self) -> AnyStr:
+    async def receive(self) -> str | bytes:
         await self.accept()
         return await self._receive()
 
-    async def send(self, data: AnyStr) -> None:
+    async def send(self, data: str | bytes) -> None:
         # Must allow for the event loop to act if the user has say
         # setup a tight loop sending data over a websocket (as in the
         # example). So yield via the sleep.