Made object streams co/contravariant (#483)

Fixes #466.

Author: agronholm

docs/versionhistory.rst

@@ -27,6 +27,8 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
     type of the value passed to ``task_status.started()``
   - The ``Listener`` class is now covariant in its stream type
   - ``create_memory_object_stream()`` now allows passing only ``item_type``
+  - Object receive streams are now covariant and object send streams are correspondingly
+    contravariant
 - Fixed ``CapacityLimiter`` on the asyncio backend to order waiting tasks in the FIFO
   order (instead of LIFO) (PR by Conor Stevenson)
 - Fixed ``CancelScope.cancel()`` not working on asyncio if called before entering the

src/anyio/abc/_streams.py

@@ -10,11 +10,12 @@
 from ._tasks import TaskGroup
 
 T_Item = TypeVar("T_Item")
-T_Stream_co = TypeVar("T_Stream_co", covariant=True)
+T_co = TypeVar("T_co", covariant=True)
+T_contra = TypeVar("T_contra", contravariant=True)
 
 
 class UnreliableObjectReceiveStream(
-    Generic[T_Item], AsyncResource, TypedAttributeProvider
+    Generic[T_co], AsyncResource, TypedAttributeProvider
 ):
     """
     An interface for receiving objects.
@@ -26,17 +27,17 @@ class UnreliableObjectReceiveStream(
     given type parameter.
     """
 
-    def __aiter__(self) -> UnreliableObjectReceiveStream[T_Item]:
+    def __aiter__(self) -> UnreliableObjectReceiveStream[T_co]:
         return self
 
-    async def __anext__(self) -> T_Item:
+    async def __anext__(self) -> T_co:
         try:
             return await self.receive()
         except EndOfStream:
             raise StopAsyncIteration
 
     @abstractmethod
-    async def receive(self) -> T_Item:
+    async def receive(self) -> T_co:
         """
         Receive the next item.
 
@@ -49,7 +50,7 @@ async def receive(self) -> T_Item:
 
 
 class UnreliableObjectSendStream(
-    Generic[T_Item], AsyncResource, TypedAttributeProvider
+    Generic[T_contra], AsyncResource, TypedAttributeProvider
 ):
     """
     An interface for sending objects.
@@ -59,7 +60,7 @@ class UnreliableObjectSendStream(
     """
 
     @abstractmethod
-    async def send(self, item: T_Item) -> None:
+    async def send(self, item: T_contra) -> None:
         """
         Send an item to the peer(s).
 
@@ -80,14 +81,14 @@ class UnreliableObjectStream(
     """
 
 
-class ObjectReceiveStream(UnreliableObjectReceiveStream[T_Item]):
+class ObjectReceiveStream(UnreliableObjectReceiveStream[T_co]):
     """
     A receive message stream which guarantees that messages are received in the same
     order in which they were sent, and that no messages are missed.
     """
 
 
-class ObjectSendStream(UnreliableObjectSendStream[T_Item]):
+class ObjectSendStream(UnreliableObjectSendStream[T_contra]):
     """
     A send message stream which guarantees that messages are delivered in the same order
     in which they were sent, without missing any messages in the middle.
@@ -186,12 +187,12 @@ async def send_eof(self) -> None:
 AnyByteStream = Union[ObjectStream[bytes], ByteStream]
 
 
-class Listener(Generic[T_Stream_co], AsyncResource, TypedAttributeProvider):
+class Listener(Generic[T_co], AsyncResource, TypedAttributeProvider):
     """An interface for objects that let you accept incoming connections."""
 
     @abstractmethod
     async def serve(
-        self, handler: Callable[[T_Stream_co], Any], task_group: TaskGroup | None = None
+        self, handler: Callable[[T_co], Any], task_group: TaskGroup | None = None
     ) -> None:
         """
         Accept incoming connections as they come in and start tasks to handle them.

src/anyio/streams/memory.py

@@ -16,6 +16,8 @@
 from ..lowlevel import checkpoint
 
 T_Item = TypeVar("T_Item")
+T_co = TypeVar("T_co", covariant=True)
+T_contra = TypeVar("T_contra", contravariant=True)
 
 
 class MemoryObjectStreamStatistics(NamedTuple):
@@ -55,14 +57,14 @@ def statistics(self) -> MemoryObjectStreamStatistics:
 
 
 @dataclass(eq=False)
-class MemoryObjectReceiveStream(Generic[T_Item], ObjectReceiveStream[T_Item]):
-    _state: MemoryObjectStreamState[T_Item]
+class MemoryObjectReceiveStream(Generic[T_co], ObjectReceiveStream[T_co]):
+    _state: MemoryObjectStreamState[T_co]
     _closed: bool = field(init=False, default=False)
 
     def __post_init__(self) -> None:
         self._state.open_receive_channels += 1
 
-    def receive_nowait(self) -> T_Item:
+    def receive_nowait(self) -> T_co:
         """
         Receive the next item if it can be done without waiting.
 
@@ -90,14 +92,14 @@ def receive_nowait(self) -> T_Item:
 
         raise WouldBlock
 
-    async def receive(self) -> T_Item:
+    async def receive(self) -> T_co:
         await checkpoint()
         try:
             return self.receive_nowait()
         except WouldBlock:
             # Add ourselves in the queue
             receive_event = Event()
-            container: list[T_Item] = []
+            container: list[T_co] = []
             self._state.waiting_receivers[receive_event] = container
 
             try:
@@ -115,7 +117,7 @@ async def receive(self) -> T_Item:
             else:
                 raise EndOfStream
 
-    def clone(self) -> MemoryObjectReceiveStream[T_Item]:
+    def clone(self) -> MemoryObjectReceiveStream[T_co]:
         """
         Create a clone of this receive stream.
 
@@ -157,7 +159,7 @@ def statistics(self) -> MemoryObjectStreamStatistics:
         """
         return self._state.statistics()
 
-    def __enter__(self) -> MemoryObjectReceiveStream[T_Item]:
+    def __enter__(self) -> MemoryObjectReceiveStream[T_co]:
         return self
 
     def __exit__(
@@ -170,14 +172,14 @@ def __exit__(
 
 
 @dataclass(eq=False)
-class MemoryObjectSendStream(Generic[T_Item], ObjectSendStream[T_Item]):
-    _state: MemoryObjectStreamState[T_Item]
+class MemoryObjectSendStream(Generic[T_contra], ObjectSendStream[T_contra]):
+    _state: MemoryObjectStreamState[T_contra]
     _closed: bool = field(init=False, default=False)
 
     def __post_init__(self) -> None:
         self._state.open_send_channels += 1
 
-    def send_nowait(self, item: T_Item) -> None:
+    def send_nowait(self, item: T_contra) -> None:
         """
         Send an item immediately if it can be done without waiting.
 
@@ -203,7 +205,7 @@ def send_nowait(self, item: T_Item) -> None:
         else:
             raise WouldBlock
 
-    async def send(self, item: T_Item) -> None:
+    async def send(self, item: T_contra) -> None:
         """
         Send an item to the stream.
 
@@ -236,7 +238,7 @@ async def send(self, item: T_Item) -> None:
             ):
                 raise BrokenResourceError
 
-    def clone(self) -> MemoryObjectSendStream[T_Item]:
+    def clone(self) -> MemoryObjectSendStream[T_contra]:
         """
         Create a clone of this send stream.
 
@@ -279,7 +281,7 @@ def statistics(self) -> MemoryObjectStreamStatistics:
         """
         return self._state.statistics()
 
-    def __enter__(self) -> MemoryObjectSendStream[T_Item]:
+    def __enter__(self) -> MemoryObjectSendStream[T_contra]:
         return self
 
     def __exit__(

tests/streams/test_memory.py

@@ -13,6 +13,7 @@
     fail_after,
     wait_all_tasks_blocked,
 )
+from anyio.abc import ObjectReceiveStream, ObjectSendStream
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 
 pytestmark = pytest.mark.anyio
@@ -345,3 +346,18 @@ async def test_sync_close() -> None:
 
     with pytest.raises(ClosedResourceError):
         receive_stream.receive_nowait()
+
+
+async def test_type_variance() -> None:
+    """
+    This test does not do anything at run time, but since the test suite is also checked
+    with a static type checker, it ensures that the memory object stream
+    co/contravariance works as intended. If it doesn't, one or both of the following
+    reassignments will trip the type checker.
+
+    """
+    send, receive = create_memory_object_stream(item_type=float)
+    receive1: MemoryObjectReceiveStream[complex] = receive  # noqa: F841
+    receive2: ObjectReceiveStream[complex] = receive  # noqa: F841
+    send1: MemoryObjectSendStream[int] = send  # noqa: F841
+    send2: ObjectSendStream[int] = send  # noqa: F841