Move exceptions into its own module

We already have quite a few exceptions, so we better move them to keep
the base classes cleaner.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/channels/__init__.py

@@ -59,19 +59,17 @@
 
 from . import util
 from ._anycast import Anycast
-from ._base_classes import (
+from ._base_classes import Peekable, Receiver, Sender
+from ._bidirectional import Bidirectional
+from ._broadcast import Broadcast
+from ._exceptions import (
     ChannelClosedError,
     ChannelError,
     Error,
-    Peekable,
-    Receiver,
     ReceiverError,
     ReceiverStoppedError,
-    Sender,
     SenderError,
 )
-from ._bidirectional import Bidirectional
-from ._broadcast import Broadcast
 
 __all__ = [
     "Anycast",

src/frequenz/channels/_anycast.py

@@ -9,11 +9,10 @@
 from collections import deque
 from typing import Deque, Generic, Optional
 
-from ._base_classes import ChannelClosedError
 from ._base_classes import Receiver as BaseReceiver
-from ._base_classes import ReceiverStoppedError
 from ._base_classes import Sender as BaseSender
-from ._base_classes import SenderError, T
+from ._base_classes import T
+from ._exceptions import ChannelClosedError, ReceiverStoppedError, SenderError
 
 
 class Anycast(Generic[T]):

src/frequenz/channels/_base_classes.py

@@ -6,105 +6,14 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import Any, Callable, Generic, Optional, TypeVar
+from typing import Callable, Generic, Optional, TypeVar
+
+from ._exceptions import ReceiverStoppedError
 
 T = TypeVar("T")
 U = TypeVar("U")
 
 
-class Error(RuntimeError):
-    """Base error.
-
-    All exceptions generated by this library inherit from this exception.
-    """
-
-    def __init__(self, message: Any):
-        """Create a ChannelError instance.
-
-        Args:
-            message: An error message.
-        """
-        super().__init__(message)
-
-
-class ChannelError(Error):
-    """An error produced in a channel.
-
-    All exceptions generated by channels inherit from this exception.
-    """
-
-    def __init__(self, message: Any, channel: Any):
-        """Create a ChannelError instance.
-
-        Args:
-            message: An error message.
-            channel: A reference to the channel that encountered the error.
-        """
-        super().__init__(message)
-        self.channel: Any = channel
-
-
-class ChannelClosedError(ChannelError):
-    """Error raised when trying to operate on a closed channel."""
-
-    def __init__(self, channel: Any):
-        """Create a `ChannelClosedError` instance.
-
-        Args:
-            channel: A reference to the channel that was closed.
-        """
-        super().__init__(f"Channel {channel} was closed", channel)
-
-
-class SenderError(Error, Generic[T]):
-    """An error produced in a [Sender][frequenz.channels.Sender].
-
-    All exceptions generated by senders inherit from this exception.
-    """
-
-    def __init__(self, message: Any, sender: Sender[T]):
-        """Create an instance.
-
-        Args:
-            message: An error message.
-            sender: The [Sender][frequenz.channels.Sender] where the error
-                happened.
-        """
-        super().__init__(message)
-        self.sender: Sender[T] = sender
-
-
-class ReceiverError(Error, Generic[T]):
-    """An error produced in a [Receiver][frequenz.channels.Receiver].
-
-    All exceptions generated by receivers inherit from this exception.
-    """
-
-    def __init__(self, message: Any, receiver: Receiver[T]):
-        """Create an instance.
-
-        Args:
-            message: An error message.
-            receiver: The [Receiver][frequenz.channels.Receiver] where the
-                error happened.
-        """
-        super().__init__(message)
-        self.receiver: Receiver[T] = receiver
-
-
-class ReceiverStoppedError(ReceiverError[T]):
-    """The [Receiver][frequenz.channels.Receiver] stopped producing messages."""
-
-    def __init__(self, receiver: Receiver[T]):
-        """Create an instance.
-
-        Args:
-            receiver: The [Receiver][frequenz.channels.Receiver] where the
-                error happened.
-        """
-        super().__init__(f"Receiver {receiver} was stopped", receiver)
-
-
 class Sender(ABC, Generic[T]):
     """A channel Sender."""
 

src/frequenz/channels/_bidirectional.py

@@ -7,16 +7,9 @@
 
 from typing import Generic, TypeVar
 
-from ._base_classes import (
-    ChannelError,
-    Receiver,
-    ReceiverError,
-    Sender,
-    SenderError,
-    T,
-    U,
-)
+from ._base_classes import Receiver, Sender, T, U
 from ._broadcast import Broadcast
+from ._exceptions import ChannelError, ReceiverError, SenderError
 
 V = TypeVar("V")
 W = TypeVar("W")

src/frequenz/channels/_broadcast.py

@@ -12,12 +12,16 @@
 from typing import Deque, Dict, Generic, Optional
 from uuid import UUID, uuid4
 
-from ._base_classes import ChannelClosedError
 from ._base_classes import Peekable as BasePeekable
 from ._base_classes import Receiver as BaseReceiver
-from ._base_classes import ReceiverError, ReceiverStoppedError
 from ._base_classes import Sender as BaseSender
-from ._base_classes import SenderError, T
+from ._base_classes import T
+from ._exceptions import (
+    ChannelClosedError,
+    ReceiverError,
+    ReceiverStoppedError,
+    SenderError,
+)
 
 logger = logging.Logger(__name__)
 

src/frequenz/channels/_exceptions.py

@@ -0,0 +1,106 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Exception classes."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Generic, TypeVar
+
+if TYPE_CHECKING:
+    from . import _base_classes
+
+T = TypeVar("T")
+
+
+class Error(RuntimeError):
+    """Base error.
+
+    All exceptions generated by this library inherit from this exception.
+    """
+
+    def __init__(self, message: Any):
+        """Create a ChannelError instance.
+
+        Args:
+            message: An error message.
+        """
+        super().__init__(message)
+
+
+class ChannelError(Error):
+    """An error produced in a channel.
+
+    All exceptions generated by channels inherit from this exception.
+    """
+
+    def __init__(self, message: Any, channel: Any):
+        """Create a ChannelError instance.
+
+        Args:
+            message: An error message.
+            channel: A reference to the channel that encountered the error.
+        """
+        super().__init__(message)
+        self.channel: Any = channel
+
+
+class ChannelClosedError(ChannelError):
+    """Error raised when trying to operate on a closed channel."""
+
+    def __init__(self, channel: Any):
+        """Create a `ChannelClosedError` instance.
+
+        Args:
+            channel: A reference to the channel that was closed.
+        """
+        super().__init__(f"Channel {channel} was closed", channel)
+
+
+class SenderError(Error, Generic[T]):
+    """An error produced in a [Sender][frequenz.channels.Sender].
+
+    All exceptions generated by senders inherit from this exception.
+    """
+
+    def __init__(self, message: Any, sender: _base_classes.Sender[T]):
+        """Create an instance.
+
+        Args:
+            message: An error message.
+            sender: The [Sender][frequenz.channels.Sender] where the error
+                happened.
+        """
+        super().__init__(message)
+        self.sender: _base_classes.Sender[T] = sender
+
+
+class ReceiverError(Error, Generic[T]):
+    """An error produced in a [Receiver][frequenz.channels.Receiver].
+
+    All exceptions generated by receivers inherit from this exception.
+    """
+
+    def __init__(self, message: Any, receiver: _base_classes.Receiver[T]):
+        """Create an instance.
+
+        Args:
+            message: An error message.
+            receiver: The [Receiver][frequenz.channels.Receiver] where the
+                error happened.
+        """
+        super().__init__(message)
+        self.receiver: _base_classes.Receiver[T] = receiver
+
+
+class ReceiverStoppedError(ReceiverError[T]):
+    """The [Receiver][frequenz.channels.Receiver] stopped producing messages."""
+
+    def __init__(self, receiver: _base_classes.Receiver[T]):
+        """Create an instance.
+
+        Args:
+            receiver: The [Receiver][frequenz.channels.Receiver] where the
+                error happened.
+        """
+        super().__init__(f"Receiver {receiver} was stopped", receiver)

src/frequenz/channels/util/_file_watcher.py

@@ -10,7 +10,8 @@
 from watchfiles import Change, awatch
 from watchfiles.main import FileChange
 
-from .._base_classes import Receiver, ReceiverStoppedError
+from .._base_classes import Receiver
+from .._exceptions import ReceiverStoppedError
 
 
 class FileWatcher(Receiver[pathlib.Path]):

src/frequenz/channels/util/_merge.py

@@ -7,7 +7,8 @@
 from collections import deque
 from typing import Any, Deque, Set
 
-from .._base_classes import Receiver, ReceiverStoppedError, T
+from .._base_classes import Receiver, T
+from .._exceptions import ReceiverStoppedError
 
 
 class Merge(Receiver[T]):

src/frequenz/channels/util/_merge_named.py

@@ -7,7 +7,8 @@
 from collections import deque
 from typing import Any, Deque, Set, Tuple
 
-from .._base_classes import Receiver, ReceiverStoppedError, T
+from .._base_classes import Receiver, T
+from .._exceptions import ReceiverStoppedError
 
 
 class MergeNamed(Receiver[Tuple[str, T]]):

src/frequenz/channels/util/_select.py

@@ -13,7 +13,8 @@
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Set, TypeVar
 
-from .._base_classes import Receiver, ReceiverStoppedError
+from .._base_classes import Receiver
+from .._exceptions import ReceiverStoppedError
 
 logger = logging.Logger(__name__)
 T = TypeVar("T")