Coverage for faststream / specification / asyncapi / v3_0_0 / schema / operations.py: 88%

1from enum import Enum (empty)

2from typing import Any (empty)

3 

4from pydantic import BaseModel, Field (empty)

5from typing_extensions import Self (empty)

6 

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

8from faststream.specification.schema.operation import Operation as OperationSpec (empty)

9 

10from .bindings import OperationBinding (empty)

11from .channels import Channel (empty)

12from .tag import Tag (empty)

13from .utils import Reference (empty)

14 

15 

16class Action(str, Enum): (empty)

17 SEND = "send" (empty)

18 RECEIVE = "receive" (empty)

19 

20 

21class Operation(BaseModel): (empty)

22 """A class to represent an operation. 

23 

24 Attributes: 

25 operation_id : ID of the operation 

26 summary : summary of the operation 

27 description : description of the operation 

28 bindings : bindings of the operation 

29 message : message of the operation 

30 security : security details of the operation 

31 tags : tags associated with the operation 

32 """ 

33 

34 action: Action (empty)

35 channel: Channel | Reference (empty)

36 

37 summary: str | None = None (empty)

38 description: str | None = None (empty)

39 

40 bindings: OperationBinding | None = None (empty)

41 

42 messages: list[Reference] = Field(default_factory=list) (empty)

43 

44 security: dict[str, list[str]] | None = None (empty)

45 

46 # TODO 

47 # traits 

48 

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

50 

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

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

53 

54 else: 

55 

56 class Config: 

57 extra = "allow" 

58 

59 @classmethod (empty)

60 def from_sub( (empty)

61 cls, 

62 messages: list[Reference], 

63 channel: Reference, 

64 operation: OperationSpec, 

65 ) -> Self: 

66 return cls( (empty)

67 action=Action.RECEIVE, 

68 messages=messages, 

69 channel=channel, 

70 bindings=OperationBinding.from_sub(operation.bindings), 

71 summary=None, 

72 description=None, 

73 security=None, 

74 tags=None, 

75 ) 

76 

77 @classmethod (empty)

78 def from_pub( (empty)

79 cls, 

80 messages: list[Reference], 

81 channel: Reference, 

82 operation: OperationSpec, 

83 ) -> Self: 

84 return cls( (empty)

85 action=Action.SEND, 

86 messages=messages, 

87 channel=channel, 

88 bindings=OperationBinding.from_pub(operation.bindings), 

89 summary=None, 

90 description=None, 

91 security=None, 

92 tags=None, 

93 )