List, Dict, Tuple -> list, dict, tuple

Author: blast-hardcheese

src/replit_river/client_transport.py

@@ -1,7 +1,7 @@
 import asyncio
 import logging
 from collections.abc import Awaitable, Callable
-from typing import Generic, Optional, Tuple
+from typing import Generic, Optional
 
 import websockets
 from pydantic import ValidationError
@@ -118,7 +118,7 @@ async def _get_existing_session(self) -> Optional[ClientSession]:
     async def _establish_new_connection(
         self,
         old_session: Optional[ClientSession] = None,
-    ) -> Tuple[
+    ) -> tuple[
         WebSocketCommonProtocol,
         ControlMessageHandshakeRequest[HandshakeMetadataType],
         ControlMessageHandshakeResponse,
@@ -292,7 +292,7 @@ async def _establish_handshake(
         handshake_metadata: HandshakeMetadataType,
         websocket: WebSocketCommonProtocol,
         old_session: Optional[ClientSession],
-    ) -> Tuple[
+    ) -> tuple[
         ControlMessageHandshakeRequest[HandshakeMetadataType],
         ControlMessageHandshakeResponse,
     ]:

src/replit_river/codegen/client.py

@@ -6,15 +6,12 @@
 from typing import (
     Any,
     Callable,
-    Dict,
-    List,
     Literal,
     Optional,
     OrderedDict,
     Sequence,
     Set,
     TextIO,
-    Tuple,
     Union,
     cast,
 )
@@ -78,8 +75,6 @@
 import datetime
 from typing import (
     Any,
-    Dict,
-    List,
     Literal,
     Optional,
     Mapping,
@@ -102,19 +97,19 @@
 
 class RiverConcreteType(BaseModel):
     type: Optional[str] = Field(default=None)
-    properties: Dict[str, "RiverType"] = Field(default_factory=lambda: dict())
+    properties: dict[str, "RiverType"] = Field(default_factory=lambda: dict())
     required: Set[str] = Field(default=set())
     items: Optional["RiverType"] = Field(default=None)
     const: Optional[Union[str, int]] = Field(default=None)
-    patternProperties: Dict[str, "RiverType"] = Field(default_factory=lambda: dict())
+    patternProperties: dict[str, "RiverType"] = Field(default_factory=lambda: dict())
 
 
 class RiverUnionType(BaseModel):
-    anyOf: List["RiverType"]
+    anyOf: list["RiverType"]
 
 
 class RiverIntersectionType(BaseModel):
-    allOf: List["RiverType"]
+    allOf: list["RiverType"]
 
 
 class RiverNotType(BaseModel):
@@ -140,11 +135,11 @@ class RiverProcedure(BaseModel):
 
 
 class RiverService(BaseModel):
-    procedures: Dict[str, RiverProcedure]
+    procedures: dict[str, RiverProcedure]
 
 
 class RiverSchema(BaseModel):
-    services: Dict[str, RiverService]
+    services: dict[str, RiverService]
     handshakeSchema: Optional[RiverConcreteType] = Field(default=None)
 
 
@@ -166,9 +161,9 @@ def encode_type(
     base_model: str,
     in_module: list[ModuleName],
     permit_unknown_members: bool,
-) -> Tuple[TypeExpression, list[ModuleName], list[FileContents], set[TypeName]]:
+) -> tuple[TypeExpression, list[ModuleName], list[FileContents], set[TypeName]]:
     encoder_name: TypeName | None = None  # defining this up here to placate mypy
-    chunks: List[FileContents] = []
+    chunks: list[FileContents] = []
     if isinstance(type, RiverNotType):
         return (NoneTypeExpr(), [], [], set())
     elif isinstance(type, RiverUnionType):
@@ -190,7 +185,7 @@ def flatten_union(tpe: RiverType) -> list[RiverType]:
 
         type = RiverUnionType(anyOf=flatten_union(type))
 
-        one_of_candidate_types: List[RiverConcreteType] = [
+        one_of_candidate_types: list[RiverConcreteType] = [
             t
             for _t in type.anyOf
             for t in (_t.anyOf if isinstance(_t, RiverUnionType) else [_t])
@@ -238,7 +233,7 @@ def flatten_union(tpe: RiverType) -> list[RiverType]:
                         (discriminator_value, []),
                     )[1].append(oneof_t)
 
-                one_of: List[TypeExpression] = []
+                one_of: list[TypeExpression] = []
                 if discriminator_name == "$kind":
                     discriminator_name = "kind"
                 for pfx, (discriminator_value, oneof_ts) in one_of_pending.items():
@@ -354,7 +349,7 @@ def {_field_name}(
             # End of stable union detection
         # Restore the non-flattened union type
         type = original_type
-        any_of: List[TypeExpression] = []
+        any_of: list[TypeExpression] = []
 
         typeddict_encoder = []
         for i, t in enumerate(type.anyOf):
@@ -521,7 +516,7 @@ def extract_props(tpe: RiverType) -> list[dict[str, RiverType]]:
             return (DictTypeExpr(type_name), module_info, type_chunks, encoder_names)
         assert type.type == "object", type.type
 
-        current_chunks: List[str] = [
+        current_chunks: list[str] = [
             f"class {render_literal_type(prefix)}({base_model}):"
         ]
         # For the encoder path, do we need "x" to be bound?
@@ -731,7 +726,7 @@ def generate_common_client(
     client_name: str,
     handshake_type: HandshakeType,
     handshake_chunks: Sequence[str],
-    modules: list[Tuple[ModuleName, ClassName]],
+    modules: list[tuple[ModuleName, ClassName]],
 ) -> FileContents:
     chunks: list[str] = [ROOT_FILE_HEADER]
     chunks.extend(
@@ -763,10 +758,10 @@ def generate_individual_service(
     schema_name: str,
     schema: RiverService,
     input_base_class: Literal["TypedDict"] | Literal["BaseModel"],
-) -> Tuple[ModuleName, ClassName, dict[RenderedPath, FileContents]]:
-    serdes: list[Tuple[list[TypeName], list[ModuleName], list[FileContents]]] = []
+) -> tuple[ModuleName, ClassName, dict[RenderedPath, FileContents]]:
+    serdes: list[tuple[list[TypeName], list[ModuleName], list[FileContents]]] = []
     class_name = ClassName(f"{schema_name.title()}Service")
-    current_chunks: List[str] = [
+    current_chunks: list[str] = [
         dedent(
             f"""\
               class {class_name}:
@@ -1124,7 +1119,7 @@ def generate_river_client_module(
     else:
         handshake_type = HandshakeType("Literal[None]")
 
-    modules: list[Tuple[ModuleName, ClassName]] = []
+    modules: list[tuple[ModuleName, ClassName]] = []
     input_base_class: Literal["TypedDict"] | Literal["BaseModel"] = (
         "TypedDict" if typed_dict_inputs else "BaseModel"
     )

src/replit_river/codegen/schema.py

@@ -4,7 +4,7 @@
 import json
 import os.path
 import tempfile
-from typing import Any, DefaultDict, Dict, List
+from typing import Any, DefaultDict
 
 import grpc_tools  # type: ignore
 from google.protobuf import descriptor_pb2
@@ -29,15 +29,15 @@ def message_type(
     module_name: str,
     m: descriptor_pb2.DescriptorProto,
     sender: bool,
-) -> Dict[str, Any]:
+) -> dict[str, Any]:
     """Generates the type of a protobuf message into Typebox descriptions."""
-    type: Dict[str, Any] = {
+    type: dict[str, Any] = {
         "type": "object",
         "properties": {},
         "required": [],
     }
     # Non-oneof fields.
-    oneofs: DefaultDict[int, List[descriptor_pb2.FieldDescriptorProto]] = (
+    oneofs: DefaultDict[int, list[descriptor_pb2.FieldDescriptorProto]] = (
         collections.defaultdict(list)
     )
     for field in m.field:
@@ -63,11 +63,11 @@ def message_type(
 def generate_river_schema(
     module_name: str,
     fds: descriptor_pb2.FileDescriptorSet,
-) -> List[Dict[str, Any]]:
+) -> list[dict[str, Any]]:
     """Generates the JSON schema of a River module."""
-    service_schemas: List[Dict[str, Any]] = []
+    service_schemas: list[dict[str, Any]] = []
 
-    message_types: Dict[str, descriptor_pb2.DescriptorProto] = {}
+    message_types: dict[str, descriptor_pb2.DescriptorProto] = {}
 
     for pd in fds.file:
         for message in pd.message_type:
@@ -80,7 +80,7 @@ def _remove_namespace(name: str) -> str:
 
         # Generate the service stubs.
         for service in pd.service:
-            service_schema: Dict[str, Any] = {
+            service_schema: dict[str, Any] = {
                 "name": "".join([service.name[0].lower(), service.name[1:]]),
                 "state": {},
                 "procedures": {},

src/replit_river/codegen/server.py

@@ -3,7 +3,7 @@
 import subprocess
 import tempfile
 from textwrap import dedent
-from typing import DefaultDict, List, Sequence
+from typing import DefaultDict, Sequence
 
 import grpc_tools  # type: ignore
 from google.protobuf import descriptor_pb2
@@ -53,7 +53,7 @@ def _{m.name}Decoder(
         ),
     ]
     # Non-oneof fields.
-    oneofs: DefaultDict[int, List[descriptor_pb2.FieldDescriptorProto]] = (
+    oneofs: DefaultDict[int, list[descriptor_pb2.FieldDescriptorProto]] = (
         collections.defaultdict(list)
     )
     for field in m.field:
@@ -224,13 +224,13 @@ def message_encoder(
             f"""\
         def _{m.name}Encoder(
             e: {module_name}_pb2.{m.name}
-        ) -> Dict[str, Any]:
-          d: Dict[str, Any] = {{}}
+        ) -> dict[str, Any]:
+          d: dict[str, Any] = {{}}
         """
         ),
     ]
     # Non-oneof fields.
-    oneofs: DefaultDict[int, List[descriptor_pb2.FieldDescriptorProto]] = (
+    oneofs: DefaultDict[int, list[descriptor_pb2.FieldDescriptorProto]] = (
         collections.defaultdict(list)
     )
     for field in m.field:
@@ -302,12 +302,12 @@ def generate_river_module(
     fds: descriptor_pb2.FileDescriptorSet,
 ) -> Sequence[str]:
     """Generates the lines of a River module."""
-    chunks: List[str] = [
+    chunks: list[str] = [
         dedent(
             f"""\
         # Code generated by river.codegen. DO NOT EDIT.
         import datetime
-        from typing import Any, Dict, Mapping, Tuple
+        from typing import Any, Mapping, Tuple
 
         from google.protobuf import timestamp_pb2
         from google.protobuf.wrappers_pb2 import BoolValue
@@ -340,8 +340,8 @@ def add_{service.name}Servicer_to_server(
                     server: river.Server,
                 ) -> None:
                   rpc_method_handlers: Mapping[
-                    Tuple[str, str],
-                    Tuple[str, river.GenericRpcHandler]
+                    tuple[str, str],
+                    tuple[str, river.GenericRpcHandler]
                   ] = {{
                 """
                 ),

src/replit_river/error_schema.py

@@ -1,4 +1,4 @@
-from typing import Any, List, Optional
+from typing import Any, Optional
 
 from pydantic import BaseModel
 
@@ -91,7 +91,7 @@ def stringify_exception(e: BaseException, limit: int = 10) -> str:
     if e.__cause__ is None:
         # If there are no causes, just fall back to stringifying the exception.
         return str(e)
-    causes: List[str] = []
+    causes: list[str] = []
     cause: Optional[BaseException] = e
     while cause and limit:
         causes.append(str(cause))

src/replit_river/rate_limiter.py

@@ -1,7 +1,6 @@
 import asyncio
 import random
 from contextvars import Context
-from typing import Dict
 
 from replit_river.transport_options import ConnectionRetryOptions
 
@@ -15,16 +14,16 @@ class LeakyBucketRateLimit:
 
     Attributes:
         options (ConnectionRetryOptions): Configuration options for retry behavior.
-        budget_consumed (Dict[str, int]): Dictionary tracking the number of retries
+        budget_consumed (dict[str, int]): Dictionary tracking the number of retries
         (or budget) consumed per user.
-        tasks (Dict[str, asyncio.Task]): Dictionary holding asyncio tasks for budget
+        tasks (dict[str, asyncio.Task]): Dictionary holding asyncio tasks for budget
         restoration.
     """
 
     def __init__(self, options: ConnectionRetryOptions):
         self.options = options
-        self.budget_consumed: Dict[str, int] = {}
-        self.tasks: Dict[str, asyncio.Task] = {}
+        self.budget_consumed: dict[str, int] = {}
+        self.tasks: dict[str, asyncio.Task] = {}
 
     def get_backoff_ms(self, user: str) -> float:
         """Calculate the backoff time in milliseconds for a user.

src/replit_river/rpc.py

@@ -6,7 +6,6 @@
     Awaitable,
     Callable,
     Coroutine,
-    Dict,
     Generic,
     Iterable,
     Iterator,
@@ -15,7 +14,6 @@
     NoReturn,
     Optional,
     Sequence,
-    Tuple,
     TypeVar,
     Union,
 )
@@ -45,7 +43,7 @@
 ResponseType = TypeVar("ResponseType")
 ErrorType = TypeVar("ErrorType", bound=RiverError)
 
-_MetadataType = Union[grpc.aio.Metadata, Sequence[Tuple[str, Union[str, bytes]]]]
+_MetadataType = Union[grpc.aio.Metadata, Sequence[tuple[str, Union[str, bytes]]]]
 
 GenericRpcHandler = Callable[
     [str, Channel[Any], Channel[Any]], Coroutine[None, None, None]
@@ -202,7 +200,7 @@ async def write(self, message: ResponseType) -> None:
 
 def get_response_or_error_payload(
     response: Any, response_serializer: Callable[[ResponseType], Any]
-) -> Dict:
+) -> dict[Any, Any]:
     if isinstance(response, RiverError):
         return {
             "ok": False,

src/replit_river/server.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from typing import Mapping, Optional, Tuple
+from typing import Mapping, Optional
 
 import websockets
 from websockets.exceptions import ConnectionClosed
@@ -35,7 +35,7 @@ async def close(self) -> None:
 
     def add_rpc_handlers(
         self,
-        rpc_handlers: Mapping[Tuple[str, str], Tuple[str, GenericRpcHandler]],
+        rpc_handlers: Mapping[tuple[str, str], tuple[str, GenericRpcHandler]],
     ) -> None:
         self._transport._handlers.update(rpc_handlers)
 

src/replit_river/server_transport.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Optional, Tuple
+from typing import Any, Optional
 
 import nanoid  # type: ignore  # type: ignore
 from pydantic import ValidationError
@@ -192,7 +192,7 @@ async def websocket_closed_callback() -> None:
 
     async def _establish_handshake(
         self, request_message: TransportMessage, websocket: WebSocketCommonProtocol
-    ) -> Tuple[
+    ) -> tuple[
         ControlMessageHandshakeRequest[Any],
         ControlMessageHandshakeResponse,
     ]: