Replace obsolete types (#176)

The tool `pyupgrade` was used to modernize the code base given that
frequenz-channels is already using python 3.11
as minimum version. The tool was mainly used to replace obsolete types.

Also `black` and `isort` tools were both run to format the code base
after running `pyupgrade`.

Related to
frequenz-floss/frequenz-sdk-python#433

Author: daniel-zullo-frequenz

benchmarks/benchmark_anycast.py

@@ -6,7 +6,8 @@
 import asyncio
 import csv
 import timeit
-from typing import Any, Coroutine, Dict, List, Tuple
+from collections.abc import Coroutine
+from typing import Any
 
 from frequenz.channels import Anycast, Receiver, Sender
 
@@ -40,7 +41,7 @@ async def benchmark_anycast(
     Returns:
         int: Total number of messages received by all channels.
     """
-    channels: List[Anycast[int]] = [Anycast(buffer_size) for _ in range(num_channels)]
+    channels: list[Anycast[int]] = [Anycast(buffer_size) for _ in range(num_channels)]
     senders = [
         asyncio.create_task(send_msg(num_messages, bcast.new_sender()))
         for bcast in channels
@@ -68,7 +69,7 @@ async def update_tracker_on_receive(chan: Receiver[int]) -> None:
     return recv_trackers[0]
 
 
-def time_async_task(task: Coroutine[Any, Any, int]) -> Tuple[float, Any]:
+def time_async_task(task: Coroutine[Any, Any, int]) -> tuple[float, Any]:
     """Run a task and return the time taken and the result.
 
     Args:
@@ -87,7 +88,7 @@ def run_one(
     num_messages: int,
     num_receivers: int,
     buffer_size: int,
-) -> Dict[str, Any]:
+) -> dict[str, Any]:
     """Run a single benchmark."""
     runtime, total_msgs = time_async_task(
         benchmark_anycast(num_channels, num_messages, num_receivers, buffer_size)

benchmarks/benchmark_broadcast.py

@@ -6,8 +6,9 @@
 import asyncio
 import csv
 import timeit
+from collections.abc import Callable, Coroutine
 from functools import partial
-from typing import Any, Callable, Coroutine, Dict, List, Tuple
+from typing import Any
 
 from frequenz.channels import Broadcast, Receiver, Sender
 
@@ -60,8 +61,8 @@ async def benchmark_broadcast(
     Returns:
         int: Total number of messages received by all receivers.
     """
-    channels: List[Broadcast[int]] = [Broadcast("meter") for _ in range(num_channels)]
-    senders: List[asyncio.Task[Any]] = [
+    channels: list[Broadcast[int]] = [Broadcast("meter") for _ in range(num_channels)]
+    senders: list[asyncio.Task[Any]] = [
         asyncio.create_task(send_msg(num_messages, bcast.new_sender()))
         for bcast in channels
     ]
@@ -103,7 +104,7 @@ async def benchmark_single_task_broadcast(
     Returns:
         int: Total number of messages received by all receivers.
     """
-    channels: List[Broadcast[int]] = [Broadcast("meter") for _ in range(num_channels)]
+    channels: list[Broadcast[int]] = [Broadcast("meter") for _ in range(num_channels)]
     senders = [b.new_sender() for b in channels]
     recv_tracker = 0
 
@@ -122,7 +123,7 @@ async def benchmark_single_task_broadcast(
     return recv_tracker
 
 
-def time_async_task(task: Coroutine[Any, Any, int]) -> Tuple[float, Any]:
+def time_async_task(task: Coroutine[Any, Any, int]) -> tuple[float, Any]:
     """Run a task and return the time taken and the result.
 
     Args:
@@ -144,7 +145,7 @@ def run_one(
     num_receivers: int,
     tasks_used: str,
     interval_between_messages: float,
-) -> Dict[str, Any]:
+) -> dict[str, Any]:
     """Run a single benchmark."""
     runtime, total_msgs = time_async_task(
         benchmark_method(num_channels, num_messages, num_receivers)

src/conftest.py

@@ -7,7 +7,6 @@
 This plugin extracts these code examples and validates them using pylint.
 """
 
-from __future__ import annotations
 
 import ast
 import os

src/frequenz/channels/_anycast.py

@@ -7,7 +7,7 @@
 
 from asyncio import Condition
 from collections import deque
-from typing import Deque, Generic, Type
+from typing import Deque, Generic
 
 from ._base_classes import Receiver as BaseReceiver
 from ._base_classes import Sender as BaseSender
@@ -169,7 +169,7 @@ def __init__(self, chan: Anycast[T]) -> None:
             chan: A reference to the channel that this receiver belongs to.
         """
         self._chan = chan
-        self._next: T | Type[_Empty] = _Empty
+        self._next: T | type[_Empty] = _Empty
 
     async def ready(self) -> bool:
         """Wait until the receiver is ready with a value or an error.

src/frequenz/channels/_base_classes.py

@@ -6,7 +6,8 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import Callable, Generic, Optional, TypeVar
+from collections.abc import Callable
+from typing import Generic, TypeVar
 
 from ._exceptions import ReceiverStoppedError
 
@@ -145,7 +146,7 @@ class Peekable(ABC, Generic[T]):
     """
 
     @abstractmethod
-    def peek(self) -> Optional[T]:
+    def peek(self) -> T | None:
         """Return the latest value that was sent to the channel.
 
         Returns:

src/frequenz/channels/_broadcast.py

@@ -9,7 +9,7 @@
 import weakref
 from asyncio import Condition
 from collections import deque
-from typing import Deque, Dict, Generic, Optional
+from typing import Deque, Generic
 from uuid import UUID, uuid4
 
 from ._base_classes import Peekable as BasePeekable
@@ -88,9 +88,9 @@ def __init__(self, name: str, resend_latest: bool = False) -> None:
         self._resend_latest = resend_latest
 
         self.recv_cv: Condition = Condition()
-        self.receivers: Dict[UUID, weakref.ReferenceType[Receiver[T]]] = {}
+        self.receivers: dict[UUID, weakref.ReferenceType[Receiver[T]]] = {}
         self.closed: bool = False
-        self._latest: Optional[T] = None
+        self._latest: T | None = None
 
     async def close(self) -> None:
         """Close the Broadcast channel.
@@ -116,9 +116,7 @@ def new_sender(self) -> Sender[T]:
         """
         return Sender(self)
 
-    def new_receiver(
-        self, name: Optional[str] = None, maxsize: int = 50
-    ) -> Receiver[T]:
+    def new_receiver(self, name: str | None = None, maxsize: int = 50) -> Receiver[T]:
         """Create a new broadcast receiver.
 
         Broadcast receivers have their own buffer, and when messages are not
@@ -346,7 +344,7 @@ def __init__(self, chan: Broadcast[T]) -> None:
         """
         self._chan = chan
 
-    def peek(self) -> Optional[T]:
+    def peek(self) -> T | None:
         """Return the latest value that was sent to the channel.
 
         Returns:

src/frequenz/channels/util/_merge.py

@@ -5,7 +5,7 @@
 
 import asyncio
 from collections import deque
-from typing import Any, Deque, Set
+from typing import Any, Deque
 
 from .._base_classes import Receiver, T
 from .._exceptions import ReceiverStoppedError
@@ -44,7 +44,7 @@ def __init__(self, *args: Receiver[T]) -> None:
             *args: sequence of channel receivers.
         """
         self._receivers = {str(id): recv for id, recv in enumerate(args)}
-        self._pending: Set[asyncio.Task[Any]] = {
+        self._pending: set[asyncio.Task[Any]] = {
             asyncio.create_task(recv.__anext__(), name=name)
             for name, recv in self._receivers.items()
         }

src/frequenz/channels/util/_merge_named.py

@@ -5,13 +5,13 @@
 
 import asyncio
 from collections import deque
-from typing import Any, Deque, Set, Tuple
+from typing import Any, Deque
 
 from .._base_classes import Receiver, T
 from .._exceptions import ReceiverStoppedError
 
 
-class MergeNamed(Receiver[Tuple[str, T]]):
+class MergeNamed(Receiver[tuple[str, T]]):
     """Merge messages coming from multiple named channels into a single stream.
 
     When `MergeNamed` is no longer needed, then it should be stopped using
@@ -25,11 +25,11 @@ def __init__(self, **kwargs: Receiver[T]) -> None:
             **kwargs: sequence of channel receivers.
         """
         self._receivers = kwargs
-        self._pending: Set[asyncio.Task[Any]] = {
+        self._pending: set[asyncio.Task[Any]] = {
             asyncio.create_task(recv.__anext__(), name=name)
             for name, recv in self._receivers.items()
         }
-        self._results: Deque[Tuple[str, T]] = deque(maxlen=len(self._receivers))
+        self._results: Deque[tuple[str, T]] = deque(maxlen=len(self._receivers))
 
     def __del__(self) -> None:
         """Cleanup any pending tasks."""
@@ -81,7 +81,7 @@ async def ready(self) -> bool:
                     asyncio.create_task(self._receivers[name].__anext__(), name=name)
                 )
 
-    def consume(self) -> Tuple[str, T]:
+    def consume(self) -> tuple[str, T]:
         """Return the latest value once `ready` is complete.
 
         Returns:

src/frequenz/channels/util/_select.py

@@ -9,7 +9,8 @@
 """
 
 import asyncio
-from typing import Any, AsyncIterator, Generic, TypeGuard, TypeVar
+from collections.abc import AsyncIterator
+from typing import Any, Generic, TypeGuard, TypeVar
 
 from .._base_classes import Receiver
 from .._exceptions import ReceiverStoppedError

tests/test_anycast.py

@@ -3,7 +3,6 @@
 
 """Tests for the Channel implementation."""
 
-from __future__ import annotations
 
 import asyncio