Coverage for faststream / specification / asyncapi / v2_6_0 / schema / message.py: 79%

1from typing import Any (empty)

2 

3from pydantic import BaseModel (empty)

4from typing_extensions import Self (empty)

5 

6from faststream._internal._compat import PYDANTIC_V2 (empty)

7from faststream.specification.asyncapi.v2_6_0.schema.tag import Tag (empty)

8from faststream.specification.schema.message import Message as SpecMessage (empty)

9 

10 

11class CorrelationId(BaseModel): (empty)

12 """A class to represent a correlation ID. 

13 

14 Attributes: 

15 description : optional description of the correlation ID 

16 location : location of the correlation ID 

17 

18 Configurations: 

19 extra : allows extra fields in the correlation ID model 

20 """ 

21 

22 location: str (empty)

23 description: str | None = None (empty)

24 

25 if PYDANTIC_V2: 25 ↛ 30line 25 didn't jump to line 30 because the condition on line 25 was always true(empty)

26 model_config = {"extra": "allow"} (empty)

27 

28 else: 

29 

30 class Config: 

31 extra = "allow" 

32 

33 

34class Message(BaseModel): (empty)

35 """A class to represent a message. 

36 

37 Attributes: 

38 title : title of the message 

39 name : name of the message 

40 summary : summary of the message 

41 description : description of the message 

42 messageId : ID of the message 

43 correlationId : correlation ID of the message 

44 contentType : content type of the message 

45 payload : dictionary representing the payload of the message 

46 tags : list of tags associated with the message 

47 """ 

48 

49 title: str | None = None (empty)

50 name: str | None = None (empty)

51 summary: str | None = None (empty)

52 description: str | None = None (empty)

53 messageId: str | None = None (empty)

54 correlationId: CorrelationId | None = None (empty)

55 contentType: str | None = None (empty)

56 

57 payload: dict[str, Any] (empty)

58 # TODO: 

59 # headers 

60 # schemaFormat 

61 # bindings 

62 # examples 

63 # traits 

64 

65 tags: list[Tag | dict[str, Any]] | None = None (empty)

66 

67 if PYDANTIC_V2: 67 ↛ 72line 67 didn't jump to line 72 because the condition on line 67 was always true(empty)

68 model_config = {"extra": "allow"} (empty)

69 

70 else: 

71 

72 class Config: 

73 extra = "allow" 

74 

75 @classmethod (empty)

76 def from_spec(cls, message: SpecMessage) -> Self: (empty)

77 return cls( (empty)

78 title=message.title, 

79 payload=message.payload, 

80 correlationId=CorrelationId( 

81 description=None, 

82 location="$message.header#/correlation_id", 

83 ), 

84 name=None, 

85 summary=None, 

86 description=None, 

87 messageId=None, 

88 contentType=None, 

89 tags=None, 

90 )