Stop using typing's Dict, List, Tuple for type annotations

We can now use plain `dict`, `list` and `tuple`.

Author: jlaine

examples/doq_server.py

@@ -2,7 +2,7 @@
 import asyncio
 import logging
 import struct
-from typing import Dict, Optional
+from typing import Optional
 
 from aioquic.asyncio import QuicConnectionProtocol, serve
 from aioquic.quic.configuration import QuicConfiguration
@@ -33,7 +33,7 @@ class SessionTicketStore:
     """
 
     def __init__(self) -> None:
-        self.tickets: Dict[bytes, SessionTicket] = {}
+        self.tickets: dict[bytes, SessionTicket] = {}
 
     def add(self, ticket: SessionTicket) -> None:
         self.tickets[ticket.ticket] = ticket

examples/http3_client.py

@@ -6,7 +6,7 @@
 import ssl
 import time
 from collections import deque
-from typing import BinaryIO, Callable, Deque, Dict, List, Optional, Union, cast
+from typing import BinaryIO, Callable, Deque, Optional, Union, cast
 from urllib.parse import urlparse
 
 import aioquic
@@ -57,7 +57,7 @@ def __init__(
         method: str,
         url: URL,
         content: bytes = b"",
-        headers: Optional[Dict] = None,
+        headers: Optional[dict] = None,
     ) -> None:
         if headers is None:
             headers = {}
@@ -125,18 +125,18 @@ class HttpClient(QuicConnectionProtocol):
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
 
-        self.pushes: Dict[int, Deque[H3Event]] = {}
+        self.pushes: dict[int, Deque[H3Event]] = {}
         self._http: Optional[HttpConnection] = None
-        self._request_events: Dict[int, Deque[H3Event]] = {}
-        self._request_waiter: Dict[int, asyncio.Future[Deque[H3Event]]] = {}
-        self._websockets: Dict[int, WebSocket] = {}
+        self._request_events: dict[int, Deque[H3Event]] = {}
+        self._request_waiter: dict[int, asyncio.Future[Deque[H3Event]]] = {}
+        self._websockets: dict[int, WebSocket] = {}
 
         if self._quic.configuration.alpn_protocols[0].startswith("hq-"):
             self._http = H0Connection(self._quic)
         else:
             self._http = H3Connection(self._quic)
 
-    async def get(self, url: str, headers: Optional[Dict] = None) -> Deque[H3Event]:
+    async def get(self, url: str, headers: Optional[dict] = None) -> Deque[H3Event]:
         """
         Perform a GET request.
         """
@@ -145,7 +145,7 @@ async def get(self, url: str, headers: Optional[Dict] = None) -> Deque[H3Event]:
         )
 
     async def post(
-        self, url: str, data: bytes, headers: Optional[Dict] = None
+        self, url: str, data: bytes, headers: Optional[dict] = None
     ) -> Deque[H3Event]:
         """
         Perform a POST request.
@@ -155,7 +155,7 @@ async def post(
         )
 
     async def websocket(
-        self, url: str, subprotocols: Optional[List[str]] = None
+        self, url: str, subprotocols: Optional[list[str]] = None
     ) -> WebSocket:
         """
         Open a WebSocket.
@@ -347,7 +347,7 @@ def save_session_ticket(ticket: SessionTicket) -> None:
 
 async def main(
     configuration: QuicConfiguration,
-    urls: List[str],
+    urls: list[str],
     data: Optional[str],
     include: bool,
     output_dir: Optional[str],

examples/http3_server.py

@@ -5,7 +5,7 @@
 import time
 from collections import deque
 from email.utils import formatdate
-from typing import Callable, Deque, Dict, List, Optional, Union, cast
+from typing import Callable, Deque, Optional, Union, cast
 
 import aioquic
 import wsproto
@@ -44,15 +44,15 @@ def __init__(
         authority: bytes,
         connection: HttpConnection,
         protocol: QuicConnectionProtocol,
-        scope: Dict,
+        scope: dict,
         stream_ended: bool,
         stream_id: int,
         transmit: Callable[[], None],
     ) -> None:
         self.authority = authority
         self.connection = connection
         self.protocol = protocol
-        self.queue: asyncio.Queue[Dict] = asyncio.Queue()
+        self.queue: asyncio.Queue[dict] = asyncio.Queue()
         self.scope = scope
         self.stream_id = stream_id
         self.transmit = transmit
@@ -77,10 +77,10 @@ def http_event_received(self, event: H3Event) -> None:
     async def run_asgi(self, app: AsgiApplication) -> None:
         await app(self.scope, self.receive, self.send)
 
-    async def receive(self) -> Dict:
+    async def receive(self) -> dict:
         return await self.queue.get()
 
-    async def send(self, message: Dict) -> None:
+    async def send(self, message: dict) -> None:
         if message["type"] == "http.response.start":
             self.connection.send_headers(
                 stream_id=self.stream_id,
@@ -129,14 +129,14 @@ def __init__(
         self,
         *,
         connection: HttpConnection,
-        scope: Dict,
+        scope: dict,
         stream_id: int,
         transmit: Callable[[], None],
     ) -> None:
         self.closed = False
         self.connection = connection
         self.http_event_queue: Deque[DataReceived] = deque()
-        self.queue: asyncio.Queue[Dict] = asyncio.Queue()
+        self.queue: asyncio.Queue[dict] = asyncio.Queue()
         self.scope = scope
         self.stream_id = stream_id
         self.transmit = transmit
@@ -171,10 +171,10 @@ async def run_asgi(self, app: AsgiApplication) -> None:
             if not self.closed:
                 await self.send({"type": "websocket.close", "code": 1000})
 
-    async def receive(self) -> Dict:
+    async def receive(self) -> dict:
         return await self.queue.get()
 
-    async def send(self, message: Dict) -> None:
+    async def send(self, message: dict) -> None:
         data = b""
         end_stream = False
         if message["type"] == "websocket.accept":
@@ -229,15 +229,15 @@ def __init__(
         self,
         *,
         connection: H3Connection,
-        scope: Dict,
+        scope: dict,
         stream_id: int,
         transmit: Callable[[], None],
     ) -> None:
         self.accepted = False
         self.closed = False
         self.connection = connection
         self.http_event_queue: Deque[H3Event] = deque()
-        self.queue: asyncio.Queue[Dict] = asyncio.Queue()
+        self.queue: asyncio.Queue[dict] = asyncio.Queue()
         self.scope = scope
         self.stream_id = stream_id
         self.transmit = transmit
@@ -274,10 +274,10 @@ async def run_asgi(self, app: AsgiApplication) -> None:
             if not self.closed:
                 await self.send({"type": "webtransport.close"})
 
-    async def receive(self) -> Dict:
+    async def receive(self) -> dict:
         return await self.queue.get()
 
-    async def send(self, message: Dict) -> None:
+    async def send(self, message: dict) -> None:
         data = b""
         end_stream = False
 
@@ -325,7 +325,7 @@ async def send(self, message: Dict) -> None:
 class HttpServerProtocol(QuicConnectionProtocol):
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
-        self._handlers: Dict[int, Handler] = {}
+        self._handlers: dict[int, Handler] = {}
         self._http: Optional[HttpConnection] = None
 
     def http_event_received(self, event: H3Event) -> None:
@@ -361,9 +361,9 @@ def http_event_received(self, event: H3Event) -> None:
             client = (client_addr[0], client_addr[1])
 
             handler: Handler
-            scope: Dict
+            scope: dict
             if method == "CONNECT" and protocol == "websocket":
-                subprotocols: List[str] = []
+                subprotocols: list[str] = []
                 for header, value in event.headers:
                     if header == b"sec-websocket-protocol":
                         subprotocols = [x.strip() for x in value.decode().split(",")]
@@ -409,7 +409,7 @@ def http_event_received(self, event: H3Event) -> None:
                     transmit=self.transmit,
                 )
             else:
-                extensions: Dict[str, Dict] = {}
+                extensions: dict[str, dict] = {}
                 if isinstance(self._http, H3Connection):
                     extensions["http.response.push"] = {}
                 scope = {
@@ -471,7 +471,7 @@ class SessionTicketStore:
     """
 
     def __init__(self) -> None:
-        self.tickets: Dict[bytes, SessionTicket] = {}
+        self.tickets: dict[bytes, SessionTicket] = {}
 
     def add(self, ticket: SessionTicket) -> None:
         self.tickets[ticket.ticket] = ticket

examples/httpx_client.py

@@ -6,7 +6,7 @@
 import ssl
 import time
 from collections import deque
-from typing import AsyncIterator, Deque, Dict, Optional, Tuple, cast
+from typing import AsyncIterator, Deque, Optional, cast
 from urllib.parse import urlparse
 
 import httpx
@@ -35,8 +35,8 @@ def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
 
         self._http = H3Connection(self._quic)
-        self._read_queue: Dict[int, Deque[H3Event]] = {}
-        self._read_ready: Dict[int, asyncio.Event] = {}
+        self._read_queue: dict[int, Deque[H3Event]] = {}
+        self._read_ready: dict[int, asyncio.Event] = {}
 
     async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
         assert isinstance(request.stream, httpx.AsyncByteStream)
@@ -94,7 +94,7 @@ def quic_event_received(self, event: QuicEvent):
             for http_event in self._http.handle_event(event):
                 self.http_event_received(http_event)
 
-    async def _receive_response(self, stream_id: int) -> Tuple[int, Headers, bool]:
+    async def _receive_response(self, stream_id: int) -> tuple[int, Headers, bool]:
         """
         Read the response status and headers.
         """

src/aioquic/_crypto.pyi

@@ -1,5 +1,3 @@
-from typing import Tuple
-
 class AEAD:
     def __init__(self, cipher_name: bytes, key: bytes, iv: bytes): ...
     def decrypt(
@@ -14,4 +12,4 @@ class CryptoError(ValueError): ...
 class HeaderProtection:
     def __init__(self, cipher_name: bytes, key: bytes): ...
     def apply(self, plain_header: bytes, protected_payload: bytes) -> bytes: ...
-    def remove(self, packet: bytes, encrypted_offset: int) -> Tuple[bytes, int]: ...
+    def remove(self, packet: bytes, encrypted_offset: int) -> tuple[bytes, int]: ...

src/aioquic/asyncio/protocol.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import Any, Callable, Dict, Optional, Text, Tuple, Union, cast
+from typing import Any, Callable, Optional, Text, Union, cast
 
 from ..quic import events
 from ..quic.connection import NetworkAddress, QuicConnection
@@ -19,9 +19,9 @@ def __init__(
         self._connected = False
         self._connected_waiter: Optional[asyncio.Future[None]] = None
         self._loop = loop
-        self._ping_waiters: Dict[int, asyncio.Future[None]] = {}
+        self._ping_waiters: dict[int, asyncio.Future[None]] = {}
         self._quic = quic
-        self._stream_readers: Dict[int, asyncio.StreamReader] = {}
+        self._stream_readers: dict[int, asyncio.StreamReader] = {}
         self._timer: Optional[asyncio.TimerHandle] = None
         self._timer_at: Optional[float] = None
         self._transmit_task: Optional[asyncio.Handle] = None
@@ -76,7 +76,7 @@ def connect(self, addr: NetworkAddress, transmit=True) -> None:
 
     async def create_stream(
         self, is_unidirectional: bool = False
-    ) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
+    ) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
         """
         Create a QUIC stream and return a pair of (reader, writer) objects.
 
@@ -182,7 +182,7 @@ def quic_event_received(self, event: events.QuicEvent) -> None:
 
     def _create_stream(
         self, stream_id: int
-    ) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
+    ) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
         adapter = QuicStreamAdapter(self, stream_id)
         reader = asyncio.StreamReader()
         protocol = asyncio.streams.StreamReaderProtocol(reader)

src/aioquic/asyncio/server.py

@@ -1,7 +1,7 @@
 import asyncio
 import os
 from functools import partial
-from typing import Callable, Dict, Optional, Text, Union, cast
+from typing import Callable, Optional, Text, Union, cast
 
 from ..buffer import Buffer
 from ..quic.configuration import SMALLEST_MAX_DATAGRAM_SIZE, QuicConfiguration
@@ -33,7 +33,7 @@ def __init__(
         self._configuration = configuration
         self._create_protocol = create_protocol
         self._loop = asyncio.get_running_loop()
-        self._protocols: Dict[bytes, QuicConnectionProtocol] = {}
+        self._protocols: dict[bytes, QuicConnectionProtocol] = {}
         self._session_ticket_fetcher = session_ticket_fetcher
         self._session_ticket_handler = session_ticket_handler
         self._transport: Optional[asyncio.DatagramTransport] = None

src/aioquic/h0/connection.py

@@ -1,5 +1,3 @@
-from typing import Dict, List
-
 from aioquic.h3.events import DataReceived, H3Event, Headers, HeadersReceived
 from aioquic.quic.connection import QuicConnection
 from aioquic.quic.events import QuicEvent, StreamDataReceived
@@ -13,13 +11,13 @@ class H0Connection:
     """
 
     def __init__(self, quic: QuicConnection):
-        self._buffer: Dict[int, bytes] = {}
-        self._headers_received: Dict[int, bool] = {}
+        self._buffer: dict[int, bytes] = {}
+        self._headers_received: dict[int, bool] = {}
         self._is_client = quic.configuration.is_client
         self._quic = quic
 
-    def handle_event(self, event: QuicEvent) -> List[H3Event]:
-        http_events: List[H3Event] = []
+    def handle_event(self, event: QuicEvent) -> list[H3Event]:
+        http_events: list[H3Event] = []
 
         if isinstance(event, StreamDataReceived) and (event.stream_id % 4) == 0:
             data = self._buffer.pop(event.stream_id, b"") + event.data