Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/anyio/abc/_streams.py: 83%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3from abc import ABCMeta, abstractmethod
4from collections.abc import Callable
5from typing import TYPE_CHECKING, Any, Generic, TypeAlias, TypeVar
7from .._core._exceptions import EndOfStream
8from .._core._typedattr import TypedAttributeProvider
9from ._resources import AsyncResource
11if TYPE_CHECKING:
12 from ._tasks import TaskGroup
14T_Item = TypeVar("T_Item")
15T_co = TypeVar("T_co", covariant=True)
16T_contra = TypeVar("T_contra", contravariant=True)
19class UnreliableObjectReceiveStream(
20 AsyncResource, TypedAttributeProvider, Generic[T_co]
21):
22 """
23 An interface for receiving objects.
25 This interface makes no guarantees that the received messages arrive in the order in
26 which they were sent, or that no messages are missed.
28 Asynchronously iterating over objects of this type will yield objects matching the
29 given type parameter.
30 """
32 def __aiter__(self) -> UnreliableObjectReceiveStream[T_co]:
33 return self
35 async def __anext__(self) -> T_co:
36 try:
37 return await self.receive()
38 except EndOfStream:
39 raise StopAsyncIteration from None
41 @abstractmethod
42 async def receive(self) -> T_co:
43 """
44 Receive the next item.
46 :raises ~anyio.ClosedResourceError: if the receive stream has been explicitly
47 closed
48 :raises ~anyio.EndOfStream: if this stream has been closed from the other end
49 :raises ~anyio.BrokenResourceError: if this stream has been rendered unusable
50 due to external causes
51 """
54class UnreliableObjectSendStream(
55 AsyncResource, TypedAttributeProvider, Generic[T_contra]
56):
57 """
58 An interface for sending objects.
60 This interface makes no guarantees that the messages sent will reach the
61 recipient(s) in the same order in which they were sent, or at all.
62 """
64 @abstractmethod
65 async def send(self, item: T_contra) -> None:
66 """
67 Send an item to the peer(s).
69 :param item: the item to send
70 :raises ~anyio.ClosedResourceError: if the send stream has been explicitly
71 closed
72 :raises ~anyio.BrokenResourceError: if this stream has been rendered unusable
73 due to external causes
74 """
77class UnreliableObjectStream(
78 UnreliableObjectReceiveStream[T_Item], UnreliableObjectSendStream[T_Item]
79):
80 """
81 A bidirectional message stream which does not guarantee the order or reliability of
82 message delivery.
83 """
86class ObjectReceiveStream(UnreliableObjectReceiveStream[T_co]):
87 """
88 A receive message stream which guarantees that messages are received in the same
89 order in which they were sent, and that no messages are missed.
90 """
93class ObjectSendStream(UnreliableObjectSendStream[T_contra]):
94 """
95 A send message stream which guarantees that messages are delivered in the same order
96 in which they were sent, without missing any messages in the middle.
97 """
100class ObjectStream(
101 ObjectReceiveStream[T_Item],
102 ObjectSendStream[T_Item],
103 UnreliableObjectStream[T_Item],
104):
105 """
106 A bidirectional message stream which guarantees the order and reliability of message
107 delivery.
108 """
110 @abstractmethod
111 async def send_eof(self) -> None:
112 """
113 Send an end-of-file indication to the peer.
115 You should not try to send any further data to this stream after calling this
116 method. This method is idempotent (does nothing on successive calls).
117 """
120class ByteReceiveStream(AsyncResource, TypedAttributeProvider):
121 """
122 An interface for receiving bytes from a single peer.
124 Iterating this byte stream will yield a byte string of arbitrary length, but no more
125 than 65536 bytes.
126 """
128 def __aiter__(self) -> ByteReceiveStream:
129 return self
131 async def __anext__(self) -> bytes:
132 try:
133 return await self.receive()
134 except EndOfStream:
135 raise StopAsyncIteration from None
137 @abstractmethod
138 async def receive(self, max_bytes: int = 65536) -> bytes:
139 """
140 Receive at most ``max_bytes`` bytes from the peer.
142 .. note:: Implementers of this interface should not return an empty
143 :class:`bytes` object, and users should ignore them.
145 :param max_bytes: maximum number of bytes to receive (must be a positive
146 integer)
147 :return: the received bytes
148 :raises ValueError: if ``max_bytes`` is less than 1
149 :raises ~anyio.EndOfStream: if this stream has been closed from the other end
150 """
153class ByteSendStream(AsyncResource, TypedAttributeProvider):
154 """An interface for sending bytes to a single peer."""
156 @abstractmethod
157 async def send(self, item: bytes) -> None:
158 """
159 Send the given bytes to the peer.
161 :param item: the bytes to send
162 """
165class ByteStream(ByteReceiveStream, ByteSendStream):
166 """A bidirectional byte stream."""
168 @abstractmethod
169 async def send_eof(self) -> None:
170 """
171 Send an end-of-file indication to the peer.
173 You should not try to send any further data to this stream after calling this
174 method. This method is idempotent (does nothing on successive calls).
175 """
178#: Type alias for all unreliable bytes-oriented receive streams.
179AnyUnreliableByteReceiveStream: TypeAlias = (
180 UnreliableObjectReceiveStream[bytes] | ByteReceiveStream
181)
182#: Type alias for all unreliable bytes-oriented send streams.
183AnyUnreliableByteSendStream: TypeAlias = (
184 UnreliableObjectSendStream[bytes] | ByteSendStream
185)
186#: Type alias for all unreliable bytes-oriented streams.
187AnyUnreliableByteStream: TypeAlias = UnreliableObjectStream[bytes] | ByteStream
188#: Type alias for all bytes-oriented receive streams.
189AnyByteReceiveStream: TypeAlias = ObjectReceiveStream[bytes] | ByteReceiveStream
190#: Type alias for all bytes-oriented send streams.
191AnyByteSendStream: TypeAlias = ObjectSendStream[bytes] | ByteSendStream
192#: Type alias for all bytes-oriented streams.
193AnyByteStream: TypeAlias = ObjectStream[bytes] | ByteStream
196class Listener(AsyncResource, TypedAttributeProvider, Generic[T_co]):
197 """An interface for objects that let you accept incoming connections."""
199 @abstractmethod
200 async def serve(
201 self, handler: Callable[[T_co], Any], task_group: TaskGroup | None = None
202 ) -> None:
203 """
204 Accept incoming connections as they come in and start tasks to handle them.
206 :param handler: a callable that will be used to handle each accepted connection
207 :param task_group: the task group that will be used to start tasks for handling
208 each accepted connection (if omitted, an ad-hoc task group will be created)
209 """
212class ObjectStreamConnectable(Generic[T_co], metaclass=ABCMeta):
213 @abstractmethod
214 async def connect(self) -> ObjectStream[T_co]:
215 """
216 Connect to the remote endpoint.
218 :return: an object stream connected to the remote end
219 :raises ConnectionFailed: if the connection fails
220 """
223class ByteStreamConnectable(metaclass=ABCMeta):
224 @abstractmethod
225 async def connect(self) -> ByteStream:
226 """
227 Connect to the remote endpoint.
229 :return: a bytestream connected to the remote end
230 :raises ConnectionFailed: if the connection fails
231 """
234#: Type alias for all connectables returning bytestreams or bytes-oriented object streams
235AnyByteStreamConnectable: TypeAlias = (
236 ObjectStreamConnectable[bytes] | ByteStreamConnectable
237)