convert events.WindowUpdated into a dataclass

Author: mhils

src/h2/connection.py

@@ -1806,9 +1806,7 @@ def _receive_window_update_frame(self, frame: WindowUpdateFrame) -> tuple[list[F
             )
 
             # FIXME: Should we split this into one event per active stream?
-            window_updated_event = WindowUpdated()
-            window_updated_event.stream_id = 0
-            window_updated_event.delta = frame.window_increment
+            window_updated_event = WindowUpdated(stream_id=0, delta=frame.window_increment)
             stream_events = [window_updated_event]
             frames = []
 

src/h2/events.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import binascii
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 from .settings import ChangedSetting, SettingCodes, Settings, _setting_code_from_int
@@ -292,6 +293,7 @@ def __repr__(self) -> str:
         )
 
 
+@dataclass(kw_only=True)
 class WindowUpdated(Event):
     """
     The WindowUpdated event is fired whenever a flow control window changes
@@ -301,13 +303,16 @@ class WindowUpdated(Event):
     the connection), and the delta in the window size.
     """
 
-    def __init__(self) -> None:
-        #: The Stream ID of the stream whose flow control window was changed.
-        #: May be ``0`` if the connection window was changed.
-        self.stream_id: int | None = None
+    stream_id: int
+    """
+    The Stream ID of the stream whose flow control window was changed.
+    May be ``0`` if the connection window was changed.
+    """
 
-        #: The window delta.
-        self.delta: int | None = None
+    delta: int | None = None
+    """
+    The window delta.
+    """
 
     def __repr__(self) -> str:
         return f"<WindowUpdated stream_id:{self.stream_id}, delta:{self.delta}>"

src/h2/stream.py

@@ -232,9 +232,7 @@ def window_updated(self, previous_state: StreamState) -> list[Event]:
         """
         Fires when a window update frame is received.
         """
-        event = WindowUpdated()
-        event.stream_id = self.stream_id
-        return [event]
+        return [WindowUpdated(stream_id=self.stream_id)]
 
     def stream_half_closed(self, previous_state: StreamState) -> list[Event]:
         """

tests/test_events.py

@@ -186,9 +186,7 @@ def test_windowupdated_repr(self) -> None:
         """
         WindowUpdated has a useful debug representation.
         """
-        e = h2.events.WindowUpdated()
-        e.stream_id = 0
-        e.delta = 2**16
+        e = h2.events.WindowUpdated(stream_id=0, delta=2**16)
 
         assert repr(e) == "<WindowUpdated stream_id:0, delta:65536>"