Use Python 3.10 unions

Author: AdrienVannson

pyproject.toml

@@ -52,6 +52,8 @@ select = [
     "SIM102", # Simplify return or yield statements
     "SIM103", # Simplify list/set/dict comprehensions
 
+    "UP007", # Use Python 3.10 unions
+
     "I",
 ]
 

src/betterproto2_compiler/enum.py

@@ -9,7 +9,6 @@
     TYPE_CHECKING,
     Any,
     Dict,
-    Optional,
     Tuple,
 )
 
@@ -111,12 +110,12 @@ class Enum(IntEnum if TYPE_CHECKING else int, metaclass=EnumType):
     inherit from this. Emulates `enum.IntEnum`.
     """
 
-    name: Optional[str]
+    name: str | None
     value: int
 
     if not TYPE_CHECKING:
 
-        def __new__(cls, *, name: Optional[str], value: int) -> Self:
+        def __new__(cls, *, name: str | None, value: int) -> Self:
             self = super().__new__(cls, value)
             super().__setattr__(self, "name", name)
             super().__setattr__(self, "value", value)

src/betterproto2_compiler/grpc/grpclib_client.py

@@ -10,7 +10,6 @@
     Optional,
     Tuple,
     Type,
-    Union,
 )
 
 import grpclib.const
@@ -25,9 +24,9 @@
     )
 
 
-Value = Union[str, bytes]
-MetadataLike = Union[Mapping[str, Value], Collection[Tuple[str, Value]]]
-MessageSource = Union[Iterable["IProtoMessage"], AsyncIterable["IProtoMessage"]]
+Value = str | bytes
+MetadataLike = Mapping[str, Value] | Collection[Tuple[str, Value]]
+MessageSource = Iterable["IProtoMessage"] | AsyncIterable["IProtoMessage"]
 
 
 class ServiceStub(ABC):
@@ -39,9 +38,9 @@ def __init__(
         self,
         channel: "Channel",
         *,
-        timeout: Optional[float] = None,
+        timeout: float | None = None,
         deadline: Optional["Deadline"] = None,
-        metadata: Optional[MetadataLike] = None,
+        metadata: MetadataLike | None = None,
     ) -> None:
         self.channel = channel
         self.timeout = timeout
@@ -50,9 +49,9 @@ def __init__(
 
     def __resolve_request_kwargs(
         self,
-        timeout: Optional[float],
+        timeout: float | None,
         deadline: Optional["Deadline"],
-        metadata: Optional[MetadataLike],
+        metadata: MetadataLike | None,
     ):
         return {
             "timeout": self.timeout if timeout is None else timeout,
@@ -66,9 +65,9 @@ async def _unary_unary(
         request: "IProtoMessage",
         response_type: Type["T"],
         *,
-        timeout: Optional[float] = None,
+        timeout: float | None = None,
         deadline: Optional["Deadline"] = None,
-        metadata: Optional[MetadataLike] = None,
+        metadata: MetadataLike | None = None,
     ) -> "T":
         """Make a unary request and return the response."""
         async with self.channel.request(
@@ -89,9 +88,9 @@ async def _unary_stream(
         request: "IProtoMessage",
         response_type: Type["T"],
         *,
-        timeout: Optional[float] = None,
+        timeout: float | None = None,
         deadline: Optional["Deadline"] = None,
-        metadata: Optional[MetadataLike] = None,
+        metadata: MetadataLike | None = None,
     ) -> AsyncIterator["T"]:
         """Make a unary request and return the stream response iterator."""
         async with self.channel.request(
@@ -112,9 +111,9 @@ async def _stream_unary(
         request_type: Type["IProtoMessage"],
         response_type: Type["T"],
         *,
-        timeout: Optional[float] = None,
+        timeout: float | None = None,
         deadline: Optional["Deadline"] = None,
-        metadata: Optional[MetadataLike] = None,
+        metadata: MetadataLike | None = None,
     ) -> "T":
         """Make a stream request and return the response."""
         async with self.channel.request(
@@ -137,9 +136,9 @@ async def _stream_stream(
         request_type: Type["IProtoMessage"],
         response_type: Type["T"],
         *,
-        timeout: Optional[float] = None,
+        timeout: float | None = None,
         deadline: Optional["Deadline"] = None,
-        metadata: Optional[MetadataLike] = None,
+        metadata: MetadataLike | None = None,
     ) -> AsyncIterator["T"]:
         """
         Make a stream request and return an AsyncIterator to iterate over response

src/betterproto2_compiler/grpc/util/async_channel.py

@@ -3,9 +3,7 @@
     AsyncIterable,
     AsyncIterator,
     Iterable,
-    Optional,
     TypeVar,
-    Union,
 )
 
 T = TypeVar("T")
@@ -117,7 +115,7 @@ def done(self) -> bool:
         # receiver per enqueued item.
         return self._closed and self._queue.qsize() <= self._waiting_receivers
 
-    async def send_from(self, source: Union[Iterable[T], AsyncIterable[T]], close: bool = False) -> "AsyncChannel[T]":
+    async def send_from(self, source: Iterable[T] | AsyncIterable[T], close: bool = False) -> "AsyncChannel[T]":
         """
         Iterates the given [Async]Iterable and sends all the resulting items.
         If close is set to True then subsequent send calls will be rejected with a
@@ -150,7 +148,7 @@ async def send(self, item: T) -> "AsyncChannel[T]":
         await self._queue.put(item)
         return self
 
-    async def receive(self) -> Optional[T]:
+    async def receive(self) -> T | None:
         """
         Returns the next item from this channel when it becomes available,
         or None if the channel is closed before another item is sent.

src/betterproto2_compiler/lib/pydantic/google/protobuf/__init__.py

@@ -40,7 +40,6 @@
     Iterable,
     Iterator,
     List,
-    Optional,
     Set,
     Type,
     Union,
@@ -416,7 +415,7 @@ def use_builtins(self) -> bool:
         )
 
     @property
-    def field_wraps(self) -> Optional[str]:
+    def field_wraps(self) -> str | None:
         """Returns betterproto wrapped field type or None."""
         match_wrapper = re.match(r"\.google\.protobuf\.(.+)Value$", self.proto_obj.type_name)
         if match_wrapper:
@@ -509,8 +508,8 @@ def betterproto_field_args(self) -> List[str]:
 
 @dataclass
 class MapEntryCompiler(FieldCompiler):
-    py_k_type: Optional[Type] = None
-    py_v_type: Optional[Type] = None
+    py_k_type: Type | None = None
+    py_v_type: Type | None = None
     proto_k_type: str = ""
     proto_v_type: str = ""
 

src/betterproto2_compiler/lib/std/google/protobuf/__init__.py

@@ -5,7 +5,6 @@
     List,
     Set,
     Tuple,
-    Union,
 )
 
 from betterproto2_compiler.lib.google.protobuf import (
@@ -45,13 +44,13 @@
 
 def traverse(
     proto_file: FileDescriptorProto,
-) -> Generator[Tuple[Union[EnumDescriptorProto, DescriptorProto], List[int]], None, None]:
+) -> Generator[Tuple[EnumDescriptorProto | DescriptorProto, List[int]], None, None]:
     # Todo: Keep information about nested hierarchy
     def _traverse(
         path: List[int],
-        items: Union[List[EnumDescriptorProto], List[DescriptorProto]],
+        items: List[EnumDescriptorProto] | List[DescriptorProto],
         prefix: str = "",
-    ) -> Generator[Tuple[Union[EnumDescriptorProto, DescriptorProto], List[int]], None, None]:
+    ) -> Generator[Tuple[EnumDescriptorProto | DescriptorProto, List[int]], None, None]:
         for i, item in enumerate(items):
             # Adjust the name since we flatten the hierarchy.
             # Todo: don't change the name, but include full name in returned tuple

src/betterproto2_compiler/plugin/models.py

@@ -7,7 +7,6 @@
 from typing import (
     Dict,
     Iterator,
-    Optional,
     Set,
 )
 
@@ -42,7 +41,7 @@ def async_iterator(self, type_: str) -> str:
         raise NotImplementedError
 
     @abc.abstractmethod
-    def imports(self) -> Dict[str, Optional[Set[str]]]:
+    def imports(self) -> Dict[str, Set[str] | None]:
         """
         Returns either the direct import as a key with none as value, or a set of
         values to import from the key.
@@ -93,7 +92,7 @@ def async_iterator(self, type_: str) -> str:
         self._imports["typing"].add("AsyncIterator")
         return f"AsyncIterator[{type_}]"
 
-    def imports(self) -> Dict[str, Optional[Set[str]]]:
+    def imports(self) -> Dict[str, Set[str] | None]:
         return {k: v if v else None for k, v in self._imports.items()}
 
 
@@ -129,7 +128,7 @@ def async_iterator(self, type_: str) -> str:
         self._imported = True
         return f"typing.AsyncIterator[{type_}]"
 
-    def imports(self) -> Dict[str, Optional[Set[str]]]:
+    def imports(self) -> Dict[str, Set[str] | None]:
         if self._imported:
             return {"typing": None}
         return {}
@@ -163,5 +162,5 @@ def async_iterator(self, type_: str) -> str:
         self._imports["collections.abc"].add("AsyncIterator")
         return f"AsyncIterator[{type_}]"
 
-    def imports(self) -> Dict[str, Optional[Set[str]]]:
+    def imports(self) -> Dict[str, Set[str] | None]:
         return {k: v if v else None for k, v in self._imports.items()}

src/betterproto2_compiler/plugin/parser.py

@@ -1,6 +1,5 @@
 from typing import (
     List,
-    Optional,
     Set,
 )
 
@@ -99,7 +98,7 @@
         ),
     ],
 )
-def test_module_validator(text: List[str], expected_collisions: Optional[Set[str]]):
+def test_module_validator(text: List[str], expected_collisions: Set[str] | None):
     line_iterator = iter(text)
     validator = ModuleValidator(line_iterator)
     valid = validator.validate()