Upgrade syntax with Ruff

Author: hugovk

docs/source/make-state-diagrams.py

@@ -25,12 +25,12 @@
 """
 
 def finish(machine_name):
-    return ("""
+    return (f"""
   labelloc="t"
   labeljust="l"
-  label=<<FONT POINT-SIZE="20">h11 state machine: {}</FONT>>
+  label=<<FONT POINT-SIZE="20">h11 state machine: {machine_name}</FONT>>
 }}
-""".format(machine_name))
+""")
 
 class Edges:
     def __init__(self):

h11/_connection.py

@@ -1,17 +1,6 @@
 # This contains the main Connection class. Everything in h11 revolves around
 # this.
-from typing import (
-    Any,
-    Callable,
-    cast,
-    Dict,
-    List,
-    Optional,
-    overload,
-    Tuple,
-    Type,
-    Union,
-)
+from typing import Any, Callable, cast, Optional, overload, Union
 
 from ._events import (
     ConnectionClosed,
@@ -92,7 +81,7 @@ def _keep_alive(event: Union[Request, Response]) -> bool:
 
 def _body_framing(
     request_method: bytes, event: Union[Request, Response]
-) -> Tuple[str, Union[Tuple[()], Tuple[int]]]:
+) -> tuple[str, Union[tuple[()], tuple[int]]]:
     # Called when we enter SEND_BODY to figure out framing information for
     # this body.
     #
@@ -166,15 +155,15 @@ class Connection:
 
     def __init__(
         self,
-        our_role: Type[Sentinel],
+        our_role: type[Sentinel],
         max_incomplete_event_size: int = DEFAULT_MAX_INCOMPLETE_EVENT_SIZE,
     ) -> None:
         self._max_incomplete_event_size = max_incomplete_event_size
         # State and role tracking
         if our_role not in (CLIENT, SERVER):
             raise ValueError(f"expected CLIENT or SERVER, not {our_role!r}")
         self.our_role = our_role
-        self.their_role: Type[Sentinel]
+        self.their_role: type[Sentinel]
         if our_role is CLIENT:
             self.their_role = SERVER
         else:
@@ -204,7 +193,7 @@ def __init__(
         self.client_is_waiting_for_100_continue = False
 
     @property
-    def states(self) -> Dict[Type[Sentinel], Type[Sentinel]]:
+    def states(self) -> dict[type[Sentinel], type[Sentinel]]:
         """A dictionary like::
 
            {CLIENT: <client state>, SERVER: <server state>}
@@ -215,14 +204,14 @@ def states(self) -> Dict[Type[Sentinel], Type[Sentinel]]:
         return dict(self._cstate.states)
 
     @property
-    def our_state(self) -> Type[Sentinel]:
+    def our_state(self) -> type[Sentinel]:
         """The current state of whichever role we are playing. See
         :ref:`state-machine` for details.
         """
         return self._cstate.states[self.our_role]
 
     @property
-    def their_state(self) -> Type[Sentinel]:
+    def their_state(self) -> type[Sentinel]:
         """The current state of whichever role we are NOT playing. See
         :ref:`state-machine` for details.
         """
@@ -252,12 +241,12 @@ def start_next_cycle(self) -> None:
         assert not self.client_is_waiting_for_100_continue
         self._respond_to_state_changes(old_states)
 
-    def _process_error(self, role: Type[Sentinel]) -> None:
+    def _process_error(self, role: type[Sentinel]) -> None:
         old_states = dict(self._cstate.states)
         self._cstate.process_error(role)
         self._respond_to_state_changes(old_states)
 
-    def _server_switch_event(self, event: Event) -> Optional[Type[Sentinel]]:
+    def _server_switch_event(self, event: Event) -> Optional[type[Sentinel]]:
         if type(event) is InformationalResponse and event.status_code == 101:
             return _SWITCH_UPGRADE
         if type(event) is Response:
@@ -269,7 +258,7 @@ def _server_switch_event(self, event: Event) -> Optional[Type[Sentinel]]:
         return None
 
     # All events go through here
-    def _process_event(self, role: Type[Sentinel], event: Event) -> None:
+    def _process_event(self, role: type[Sentinel], event: Event) -> None:
         # First, pass the event through the state machine to make sure it
         # succeeds.
         old_states = dict(self._cstate.states)
@@ -319,7 +308,7 @@ def _process_event(self, role: Type[Sentinel], event: Event) -> None:
 
     def _get_io_object(
         self,
-        role: Type[Sentinel],
+        role: type[Sentinel],
         event: Optional[Event],
         io_dict: Union[ReadersType, WritersType],
     ) -> Optional[Callable[..., Any]]:
@@ -341,7 +330,7 @@ def _get_io_object(
     # self._cstate.states to change.
     def _respond_to_state_changes(
         self,
-        old_states: Dict[Type[Sentinel], Type[Sentinel]],
+        old_states: dict[type[Sentinel], type[Sentinel]],
         event: Optional[Event] = None,
     ) -> None:
         # Update reader/writer
@@ -351,7 +340,7 @@ def _respond_to_state_changes(
             self._reader = self._get_io_object(self.their_role, event, READERS)
 
     @property
-    def trailing_data(self) -> Tuple[bytes, bool]:
+    def trailing_data(self) -> tuple[bytes, bool]:
         """Data that has been received, but not yet processed, represented as
         a tuple with two elements, where the first is a byte-string containing
         the unprocessed data itself, and the second is a bool that is True if
@@ -409,7 +398,7 @@ def receive_data(self, data: bytes) -> None:
 
     def _extract_next_receive_event(
         self,
-    ) -> Union[Event, Type[NEED_DATA], Type[PAUSED]]:
+    ) -> Union[Event, type[NEED_DATA], type[PAUSED]]:
         state = self.their_state
         # We don't pause immediately when they enter DONE, because even in
         # DONE state we can still process a ConnectionClosed() event. But
@@ -435,7 +424,7 @@ def _extract_next_receive_event(
             event = NEED_DATA
         return event  # type: ignore[no-any-return]
 
-    def next_event(self) -> Union[Event, Type[NEED_DATA], Type[PAUSED]]:
+    def next_event(self) -> Union[Event, type[NEED_DATA], type[PAUSED]]:
         """Parse the next event out of our receive buffer, update our internal
         state, and return it.
 
@@ -541,7 +530,7 @@ def send(self, event: Event) -> Optional[bytes]:
         else:
             return b"".join(data_list)
 
-    def send_with_data_passthrough(self, event: Event) -> Optional[List[bytes]]:
+    def send_with_data_passthrough(self, event: Event) -> Optional[list[bytes]]:
         """Identical to :meth:`send`, except that in situations where
         :meth:`send` returns a single :term:`bytes-like object`, this instead
         returns a list of them -- and when sending a :class:`Data` event, this
@@ -567,7 +556,7 @@ def send_with_data_passthrough(self, event: Event) -> Optional[List[bytes]]:
                 # In any situation where writer is None, process_event should
                 # have raised ProtocolError
                 assert writer is not None
-                data_list: List[bytes] = []
+                data_list: list[bytes] = []
                 writer(event, data_list.append)
                 return data_list
         except:

h11/_events.py

@@ -8,7 +8,7 @@
 import re
 from abc import ABC
 from dataclasses import dataclass
-from typing import List, Tuple, Union
+from typing import Union
 
 from ._abnf import method, request_target
 from ._headers import Headers, normalize_and_validate
@@ -83,7 +83,7 @@ def __init__(
         self,
         *,
         method: Union[bytes, str],
-        headers: Union[Headers, List[Tuple[bytes, bytes]], List[Tuple[str, str]]],
+        headers: Union[Headers, list[tuple[bytes, bytes]], list[tuple[str, str]]],
         target: Union[bytes, str],
         http_version: Union[bytes, str] = b"1.1",
         _parsed: bool = False,
@@ -137,7 +137,7 @@ class _ResponseBase(Event):
     def __init__(
         self,
         *,
-        headers: Union[Headers, List[Tuple[bytes, bytes]], List[Tuple[str, str]]],
+        headers: Union[Headers, list[tuple[bytes, bytes]], list[tuple[str, str]]],
         status_code: int,
         http_version: Union[bytes, str] = b"1.1",
         reason: Union[bytes, str] = b"",
@@ -207,7 +207,7 @@ def __post_init__(self) -> None:
         if not (100 <= self.status_code < 200):
             raise LocalProtocolError(
                 "InformationalResponse status_code should be in range "
-                "[100, 200), not {}".format(self.status_code)
+                f"[100, 200), not {self.status_code}"
             )
 
     # This is an unhashable type.
@@ -247,9 +247,7 @@ class Response(_ResponseBase):
     def __post_init__(self) -> None:
         if not (200 <= self.status_code < 1000):
             raise LocalProtocolError(
-                "Response status_code should be in range [200, 1000), not {}".format(
-                    self.status_code
-                )
+                f"Response status_code should be in range [200, 1000), not {self.status_code}"
             )
 
     # This is an unhashable type.
@@ -338,7 +336,7 @@ def __init__(
         self,
         *,
         headers: Union[
-            Headers, List[Tuple[bytes, bytes]], List[Tuple[str, str]], None
+            Headers, list[tuple[bytes, bytes]], list[tuple[str, str]], None
         ] = None,
         _parsed: bool = False,
     ) -> None:

h11/_headers.py

@@ -1,5 +1,6 @@
 import re
-from typing import AnyStr, cast, List, overload, Sequence, Tuple, TYPE_CHECKING, Union
+from collections.abc import Sequence
+from typing import overload, TYPE_CHECKING, Union
 
 from ._abnf import field_name, field_value
 from ._util import bytesify, LocalProtocolError, validate
@@ -74,7 +75,7 @@
 _field_value_re = re.compile(field_value.encode("ascii"))
 
 
-class Headers(Sequence[Tuple[bytes, bytes]]):
+class Headers(Sequence[tuple[bytes, bytes]]):
     """
     A list-like interface that allows iterating over headers as byte-pairs
     of (lowercased-name, value).
@@ -101,7 +102,7 @@ class Headers(Sequence[Tuple[bytes, bytes]]):
 
     __slots__ = "_full_items"
 
-    def __init__(self, full_items: List[Tuple[bytes, bytes, bytes]]) -> None:
+    def __init__(self, full_items: list[tuple[bytes, bytes, bytes]]) -> None:
         self._full_items = full_items
 
     def __bool__(self) -> bool:
@@ -114,21 +115,21 @@ def __len__(self) -> int:
         return len(self._full_items)
 
     def __repr__(self) -> str:
-        return "<Headers(%s)>" % repr(list(self))
+        return f"<Headers({repr(list(self))})>"
 
-    def __getitem__(self, idx: int) -> Tuple[bytes, bytes]:  # type: ignore[override]
+    def __getitem__(self, idx: int) -> tuple[bytes, bytes]:  # type: ignore[override]
         _, name, value = self._full_items[idx]
         return (name, value)
 
-    def raw_items(self) -> List[Tuple[bytes, bytes]]:
+    def raw_items(self) -> list[tuple[bytes, bytes]]:
         return [(raw_name, value) for raw_name, _, value in self._full_items]
 
 
 HeaderTypes = Union[
-    List[Tuple[bytes, bytes]],
-    List[Tuple[bytes, str]],
-    List[Tuple[str, bytes]],
-    List[Tuple[str, str]],
+    list[tuple[bytes, bytes]],
+    list[tuple[bytes, str]],
+    list[tuple[str, bytes]],
+    list[tuple[str, str]],
 ]
 
 
@@ -206,7 +207,7 @@ def normalize_and_validate(
     return Headers(new_headers)
 
 
-def get_comma_header(headers: Headers, name: bytes) -> List[bytes]:
+def get_comma_header(headers: Headers, name: bytes) -> list[bytes]:
     # Should only be used for headers whose value is a list of
     # comma-separated, case-insensitive values.
     #
@@ -242,7 +243,7 @@ def get_comma_header(headers: Headers, name: bytes) -> List[bytes]:
     # Expect: the only legal value is the literal string
     # "100-continue". Splitting on commas is harmless. Case insensitive.
     #
-    out: List[bytes] = []
+    out: list[bytes] = []
     for _, found_name, found_raw_value in headers._full_items:
         if found_name == name:
             found_raw_value = found_raw_value.lower()
@@ -253,7 +254,7 @@ def get_comma_header(headers: Headers, name: bytes) -> List[bytes]:
     return out
 
 
-def set_comma_header(headers: Headers, name: bytes, new_values: List[bytes]) -> Headers:
+def set_comma_header(headers: Headers, name: bytes, new_values: list[bytes]) -> Headers:
     # The header name `name` is expected to be lower-case bytes.
     #
     # Note that when we store the header we use title casing for the header
@@ -263,7 +264,7 @@ def set_comma_header(headers: Headers, name: bytes, new_values: List[bytes]) ->
     # here given the cases where we're using `set_comma_header`...
     #
     # Connection, Content-Length, Transfer-Encoding.
-    new_headers: List[Tuple[bytes, bytes]] = []
+    new_headers: list[tuple[bytes, bytes]] = []
     for found_raw_name, found_name, found_raw_value in headers._full_items:
         if found_name != name:
             new_headers.append((found_raw_name, found_raw_value))

h11/_readers.py

@@ -17,7 +17,8 @@
 # - or, for body readers, a dict of per-framing reader factories
 
 import re
-from typing import Any, Callable, Dict, Iterable, NoReturn, Optional, Tuple, Type, Union
+from collections.abc import Iterable
+from typing import Any, Callable, NoReturn, Optional, Union
 
 from ._abnf import chunk_header, header_field, request_line, status_line
 from ._events import Data, EndOfMessage, InformationalResponse, Request, Response
@@ -63,7 +64,7 @@ def _obsolete_line_fold(lines: Iterable[bytes]) -> Iterable[bytes]:
 
 def _decode_header_lines(
     lines: Iterable[bytes],
-) -> Iterable[Tuple[bytes, bytes]]:
+) -> Iterable[tuple[bytes, bytes]]:
     for line in _obsolete_line_fold(lines):
         matches = validate(header_field_re, line, "illegal header line: {!r}", line)
         yield (matches["field_name"], matches["field_value"])
@@ -107,7 +108,7 @@ def maybe_read_from_SEND_RESPONSE_server(
     )
     reason = b"" if matches["reason"] is None else matches["reason"]
     status_code = int(matches["status_code"])
-    class_: Union[Type[InformationalResponse], Type[Response]] = (
+    class_: Union[type[InformationalResponse], type[Response]] = (
         InformationalResponse if status_code < 200 else Response
     )
     return class_(
@@ -136,9 +137,7 @@ def __call__(self, buf: ReceiveBuffer) -> Union[Data, EndOfMessage, None]:
     def read_eof(self) -> NoReturn:
         raise RemoteProtocolError(
             "peer closed connection without sending complete message body "
-            "(received {} bytes, expected {})".format(
-                self._length - self._remaining, self._length
-            )
+            f"(received {self._length - self._remaining} bytes, expected {self._length})"
         )
 
 
@@ -227,9 +226,9 @@ def expect_nothing(buf: ReceiveBuffer) -> None:
     return None
 
 
-ReadersType = Dict[
-    Union[Type[Sentinel], Tuple[Type[Sentinel], Type[Sentinel]]],
-    Union[Callable[..., Any], Dict[str, Callable[..., Any]]],
+ReadersType = dict[
+    Union[type[Sentinel], tuple[type[Sentinel], type[Sentinel]]],
+    Union[Callable[..., Any], dict[str, Callable[..., Any]]],
 ]
 
 READERS: ReadersType = {

h11/_receivebuffer.py

@@ -1,6 +1,5 @@
 import re
-import sys
-from typing import List, Optional, Union
+from typing import Optional, Union
 
 __all__ = ["ReceiveBuffer"]
 
@@ -101,7 +100,7 @@ def maybe_extract_next_line(self) -> Optional[bytearray]:
 
         return self._extract(idx)
 
-    def maybe_extract_lines(self) -> Optional[List[bytearray]]:
+    def maybe_extract_lines(self) -> Optional[list[bytearray]]:
         """
         Extract everything up to the first blank line, and return a list of lines.
         """