Switch from typing.Optional to | None.

Author: aaugustin

src/websockets/client.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import warnings
-from typing import Any, Generator, List, Optional, Sequence
+from typing import Any, Generator, List, Sequence
 
 from .datastructures import Headers, MultipleValuesError
 from .exceptions import (
@@ -73,12 +73,12 @@ def __init__(
         self,
         wsuri: WebSocketURI,
         *,
-        origin: Optional[Origin] = None,
-        extensions: Optional[Sequence[ClientExtensionFactory]] = None,
-        subprotocols: Optional[Sequence[Subprotocol]] = None,
+        origin: Origin | None = None,
+        extensions: Sequence[ClientExtensionFactory] | None = None,
+        subprotocols: Sequence[Subprotocol] | None = None,
         state: State = CONNECTING,
-        max_size: Optional[int] = 2**20,
-        logger: Optional[LoggerLike] = None,
+        max_size: int | None = 2**20,
+        logger: LoggerLike | None = None,
     ):
         super().__init__(
             side=CLIENT,
@@ -261,7 +261,7 @@ def process_extensions(self, headers: Headers) -> List[Extension]:
 
         return accepted_extensions
 
-    def process_subprotocol(self, headers: Headers) -> Optional[Subprotocol]:
+    def process_subprotocol(self, headers: Headers) -> Subprotocol | None:
         """
         Handle the Sec-WebSocket-Protocol HTTP response header.
 
@@ -274,7 +274,7 @@ def process_subprotocol(self, headers: Headers) -> Optional[Subprotocol]:
            Subprotocol, if one was selected.
 
         """
-        subprotocol: Optional[Subprotocol] = None
+        subprotocol: Subprotocol | None = None
 
         subprotocols = headers.get_all("Sec-WebSocket-Protocol")
 

src/websockets/exceptions.py

@@ -31,7 +31,6 @@
 from __future__ import annotations
 
 import http
-from typing import Optional
 
 from . import datastructures, frames, http11
 from .typing import StatusLike
@@ -78,21 +77,21 @@ class ConnectionClosed(WebSocketException):
     Raised when trying to interact with a closed connection.
 
     Attributes:
-        rcvd (Optional[Close]): if a close frame was received, its code and
+        rcvd (Close | None): if a close frame was received, its code and
             reason are available in ``rcvd.code`` and ``rcvd.reason``.
-        sent (Optional[Close]): if a close frame was sent, its code and reason
+        sent (Close | None): if a close frame was sent, its code and reason
             are available in ``sent.code`` and ``sent.reason``.
-        rcvd_then_sent (Optional[bool]): if close frames were received and
+        rcvd_then_sent (bool | None): if close frames were received and
             sent, this attribute tells in which order this happened, from the
             perspective of this side of the connection.
 
     """
 
     def __init__(
         self,
-        rcvd: Optional[frames.Close],
-        sent: Optional[frames.Close],
-        rcvd_then_sent: Optional[bool] = None,
+        rcvd: frames.Close | None,
+        sent: frames.Close | None,
+        rcvd_then_sent: bool | None = None,
     ) -> None:
         self.rcvd = rcvd
         self.sent = sent
@@ -181,7 +180,7 @@ class InvalidHeader(InvalidHandshake):
 
     """
 
-    def __init__(self, name: str, value: Optional[str] = None) -> None:
+    def __init__(self, name: str, value: str | None = None) -> None:
         self.name = name
         self.value = value
 
@@ -221,7 +220,7 @@ class InvalidOrigin(InvalidHeader):
 
     """
 
-    def __init__(self, origin: Optional[str]) -> None:
+    def __init__(self, origin: str | None) -> None:
         super().__init__("Origin", origin)
 
 
@@ -301,7 +300,7 @@ class InvalidParameterValue(NegotiationError):
 
     """
 
-    def __init__(self, name: str, value: Optional[str]) -> None:
+    def __init__(self, name: str, value: str | None) -> None:
         self.name = name
         self.value = value
 

src/websockets/extensions/base.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import List, Optional, Sequence, Tuple
+from typing import List, Sequence, Tuple
 
 from .. import frames
 from ..typing import ExtensionName, ExtensionParameter
@@ -22,7 +22,7 @@ def decode(
         self,
         frame: frames.Frame,
         *,
-        max_size: Optional[int] = None,
+        max_size: int | None = None,
     ) -> frames.Frame:
         """
         Decode an incoming frame.

src/websockets/extensions/permessage_deflate.py

@@ -2,7 +2,7 @@
 
 import dataclasses
 import zlib
-from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
+from typing import Any, Dict, List, Sequence, Tuple, Union
 
 from .. import exceptions, frames
 from ..typing import ExtensionName, ExtensionParameter
@@ -36,7 +36,7 @@ def __init__(
         local_no_context_takeover: bool,
         remote_max_window_bits: int,
         local_max_window_bits: int,
-        compress_settings: Optional[Dict[Any, Any]] = None,
+        compress_settings: Dict[Any, Any] | None = None,
     ) -> None:
         """
         Configure the Per-Message Deflate extension.
@@ -84,7 +84,7 @@ def decode(
         self,
         frame: frames.Frame,
         *,
-        max_size: Optional[int] = None,
+        max_size: int | None = None,
     ) -> frames.Frame:
         """
         Decode an incoming frame.
@@ -174,8 +174,8 @@ def encode(self, frame: frames.Frame) -> frames.Frame:
 def _build_parameters(
     server_no_context_takeover: bool,
     client_no_context_takeover: bool,
-    server_max_window_bits: Optional[int],
-    client_max_window_bits: Optional[Union[int, bool]],
+    server_max_window_bits: int | None,
+    client_max_window_bits: Union[int, bool] | None,
 ) -> List[ExtensionParameter]:
     """
     Build a list of ``(name, value)`` pairs for some compression parameters.
@@ -197,7 +197,7 @@ def _build_parameters(
 
 def _extract_parameters(
     params: Sequence[ExtensionParameter], *, is_server: bool
-) -> Tuple[bool, bool, Optional[int], Optional[Union[int, bool]]]:
+) -> Tuple[bool, bool, int | None, Union[int, bool] | None]:
     """
     Extract compression parameters from a list of ``(name, value)`` pairs.
 
@@ -207,8 +207,8 @@ def _extract_parameters(
     """
     server_no_context_takeover: bool = False
     client_no_context_takeover: bool = False
-    server_max_window_bits: Optional[int] = None
-    client_max_window_bits: Optional[Union[int, bool]] = None
+    server_max_window_bits: int | None = None
+    client_max_window_bits: Union[int, bool] | None = None
 
     for name, value in params:
         if name == "server_no_context_takeover":
@@ -286,9 +286,9 @@ def __init__(
         self,
         server_no_context_takeover: bool = False,
         client_no_context_takeover: bool = False,
-        server_max_window_bits: Optional[int] = None,
-        client_max_window_bits: Optional[Union[int, bool]] = True,
-        compress_settings: Optional[Dict[str, Any]] = None,
+        server_max_window_bits: int | None = None,
+        client_max_window_bits: Union[int, bool] | None = True,
+        compress_settings: Dict[str, Any] | None = None,
     ) -> None:
         """
         Configure the Per-Message Deflate extension factory.
@@ -433,7 +433,7 @@ def process_response_params(
 
 
 def enable_client_permessage_deflate(
-    extensions: Optional[Sequence[ClientExtensionFactory]],
+    extensions: Sequence[ClientExtensionFactory] | None,
 ) -> Sequence[ClientExtensionFactory]:
     """
     Enable Per-Message Deflate with default settings in client extensions.
@@ -489,9 +489,9 @@ def __init__(
         self,
         server_no_context_takeover: bool = False,
         client_no_context_takeover: bool = False,
-        server_max_window_bits: Optional[int] = None,
-        client_max_window_bits: Optional[int] = None,
-        compress_settings: Optional[Dict[str, Any]] = None,
+        server_max_window_bits: int | None = None,
+        client_max_window_bits: int | None = None,
+        compress_settings: Dict[str, Any] | None = None,
         require_client_max_window_bits: bool = False,
     ) -> None:
         """
@@ -635,7 +635,7 @@ def process_request_params(
 
 
 def enable_server_permessage_deflate(
-    extensions: Optional[Sequence[ServerExtensionFactory]],
+    extensions: Sequence[ServerExtensionFactory] | None,
 ) -> Sequence[ServerExtensionFactory]:
     """
     Enable Per-Message Deflate with default settings in server extensions.

src/websockets/frames.py

@@ -5,7 +5,7 @@
 import io
 import secrets
 import struct
-from typing import Callable, Generator, Optional, Sequence, Tuple
+from typing import Callable, Generator, Sequence, Tuple
 
 from . import exceptions, extensions
 from .typing import Data
@@ -205,8 +205,8 @@ def parse(
         read_exact: Callable[[int], Generator[None, None, bytes]],
         *,
         mask: bool,
-        max_size: Optional[int] = None,
-        extensions: Optional[Sequence[extensions.Extension]] = None,
+        max_size: int | None = None,
+        extensions: Sequence[extensions.Extension] | None = None,
     ) -> Generator[None, None, Frame]:
         """
         Parse a WebSocket frame.
@@ -280,7 +280,7 @@ def serialize(
         self,
         *,
         mask: bool,
-        extensions: Optional[Sequence[extensions.Extension]] = None,
+        extensions: Sequence[extensions.Extension] | None = None,
     ) -> bytes:
         """
         Serialize a WebSocket frame.

src/websockets/headers.py

@@ -4,7 +4,7 @@
 import binascii
 import ipaddress
 import re
-from typing import Callable, List, Optional, Sequence, Tuple, TypeVar, cast
+from typing import Callable, List, Sequence, Tuple, TypeVar, cast
 
 from . import exceptions
 from .typing import (
@@ -63,7 +63,7 @@ def build_host(host: str, port: int, secure: bool) -> str:
 # https://www.rfc-editor.org/rfc/rfc7230.html#appendix-B.
 
 
-def peek_ahead(header: str, pos: int) -> Optional[str]:
+def peek_ahead(header: str, pos: int) -> str | None:
     """
     Return the next character from ``header`` at the given position.
 
@@ -314,7 +314,7 @@ def parse_extension_item_param(
     name, pos = parse_token(header, pos, header_name)
     pos = parse_OWS(header, pos)
     # Extract parameter value, if there is one.
-    value: Optional[str] = None
+    value: str | None = None
     if peek_ahead(header, pos) == "=":
         pos = parse_OWS(header, pos + 1)
         if peek_ahead(header, pos) == '"':

src/websockets/http11.py

@@ -3,7 +3,7 @@
 import dataclasses
 import re
 import warnings
-from typing import Callable, Generator, Optional
+from typing import Callable, Generator
 
 from . import datastructures, exceptions
 
@@ -62,10 +62,10 @@ class Request:
     headers: datastructures.Headers
     # body isn't useful is the context of this library.
 
-    _exception: Optional[Exception] = None
+    _exception: Exception | None = None
 
     @property
-    def exception(self) -> Optional[Exception]:  # pragma: no cover
+    def exception(self) -> Exception | None:  # pragma: no cover
         warnings.warn(
             "Request.exception is deprecated; "
             "use ServerProtocol.handshake_exc instead",
@@ -164,12 +164,12 @@ class Response:
     status_code: int
     reason_phrase: str
     headers: datastructures.Headers
-    body: Optional[bytes] = None
+    body: bytes | None = None
 
-    _exception: Optional[Exception] = None
+    _exception: Exception | None = None
 
     @property
-    def exception(self) -> Optional[Exception]:  # pragma: no cover
+    def exception(self) -> Exception | None:  # pragma: no cover
         warnings.warn(
             "Response.exception is deprecated; "
             "use ClientProtocol.handshake_exc instead",
@@ -245,7 +245,7 @@ def parse(
         if 100 <= status_code < 200 or status_code == 204 or status_code == 304:
             body = None
         else:
-            content_length: Optional[int]
+            content_length: int | None
             try:
                 # MultipleValuesError is sufficiently unlikely that we don't
                 # attempt to handle it. Instead we document that its parent

src/websockets/imports.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import warnings
-from typing import Any, Dict, Iterable, Optional
+from typing import Any, Dict, Iterable
 
 
 __all__ = ["lazy_import"]
@@ -30,8 +30,8 @@ def import_name(name: str, source: str, namespace: Dict[str, Any]) -> Any:
 
 def lazy_import(
     namespace: Dict[str, Any],
-    aliases: Optional[Dict[str, str]] = None,
-    deprecated_aliases: Optional[Dict[str, str]] = None,
+    aliases: Dict[str, str] | None = None,
+    deprecated_aliases: Dict[str, str] | None = None,
 ) -> None:
     """
     Provide lazy, module-level imports.