Coverage for faststream / _internal / parser.py: 91%

1from abc import abstractmethod (empty)

2from collections.abc import Sequence (empty)

3from typing import TYPE_CHECKING, Any, Protocol, TypeVar (empty)

4 

5from faststream.message.utils import decode_message, encode_message (empty)

6 

7if TYPE_CHECKING: (empty)

8 from fast_depends.library.serializer import SerializerProto 

9 

10 from faststream._internal.basic_types import DecodedMessage, SendableMessage 

11 from faststream.message import StreamMessage 

12 

13MsgType = TypeVar("MsgType") (empty)

14 

15 

16class ParserProto(Protocol[MsgType]): (empty)

17 """Protocol for parsing raw messages into StreamMessage.""" 

18 

19 @abstractmethod (empty)

20 async def parse_message(self, message: MsgType) -> "StreamMessage[MsgType]": (empty)

21 """Parse a raw message into a StreamMessage.""" 

22 ... 

23 

24 

25class DecoderProto(Protocol): (empty)

26 """Protocol for decoding StreamMessage into DecodedMessage.""" 

27 

28 @abstractmethod (empty)

29 async def decode_message(self, msg: "StreamMessage[Any]") -> "DecodedMessage": (empty)

30 """Decode a StreamMessage into a DecodedMessage.""" 

31 ... 

32 

33 

34class BatchParserProto(Protocol[MsgType]): (empty)

35 """Protocol for parsing a batch of raw messages into StreamMessage.""" 

36 

37 @abstractmethod (empty)

38 async def parse_batch( (empty)

39 self, messages: Sequence[MsgType] (empty)

40 ) -> "StreamMessage[Sequence[MsgType]]": (empty)

41 """Parse a batch of raw messages into a StreamMessage.""" 

42 ... 

43 

44 

45class BatchDecoderProto(Protocol[MsgType]): (empty)

46 """Protocol for decoding a batch of StreamMessage into list of DecodedMessage.""" 

47 

48 @abstractmethod (empty)

49 async def decode_batch( (empty)

50 self, msg: "StreamMessage[Sequence[MsgType]]" (empty)

51 ) -> list["DecodedMessage"]: (empty)

52 """Decode a batch of StreamMessage into a list of DecodedMessage.""" 

53 ... 

54 

55 

56class CodecProto(Protocol): (empty)

57 """Protocol for encoding and decoding message bodies.""" 

58 

59 @abstractmethod (empty)

60 async def decode(self, msg: "StreamMessage[Any]") -> "DecodedMessage": (empty)

61 """Decode a StreamMessage body into a Python object.""" 

62 ... 

63 

64 @abstractmethod (empty)

65 async def encode( (empty)

66 self, 

67 msg: "SendableMessage", (empty)

68 serializer: "SerializerProto | None" = None, (empty)

69 ) -> tuple[bytes, str | None]: (empty)

70 """Encode a Python object into bytes and content_type.""" 

71 ... 

72 

73 

74class DefaultCodec: (empty)

75 """Default codec that delegates to the shared encode_message/decode_message functions.""" 

76 

77 async def decode(self, msg: "StreamMessage[Any]") -> "DecodedMessage": (empty)

78 return decode_message(msg) (empty)

79 

80 async def encode( (empty)

81 self, 

82 msg: "SendableMessage", 

83 serializer: "SerializerProto | None" = None, 

84 ) -> tuple[bytes, str | None]: 

85 return encode_message(msg, serializer)