Coverage for faststream / specification / asyncapi / v2_6_0 / schema / bindings / main / channel.py: 75%

1from typing import overload (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.bindings import ( (empty)

8 amqp as amqp_bindings, 

9 kafka as kafka_bindings, 

10 mqtt as mqtt_bindings, 

11 nats as nats_bindings, 

12 redis as redis_bindings, 

13 sqs as sqs_bindings, 

14) 

15from faststream.specification.schema.bindings import ChannelBinding as SpecBinding (empty)

16 

17 

18class ChannelBinding(BaseModel): (empty)

19 """A class to represent channel bindings. 

20 

21 Attributes: 

22 amqp : AMQP channel binding (optional) 

23 kafka : Kafka channel binding (optional) 

24 sqs : SQS channel binding (optional) 

25 nats : NATS channel binding (optional) 

26 redis : Redis channel binding (optional) 

27 """ 

28 

29 amqp: amqp_bindings.ChannelBinding | None = None (empty)

30 kafka: kafka_bindings.ChannelBinding | None = None (empty)

31 mqtt: mqtt_bindings.ChannelBinding | None = None (empty)

32 sqs: sqs_bindings.ChannelBinding | None = None (empty)

33 nats: nats_bindings.ChannelBinding | None = None (empty)

34 redis: redis_bindings.ChannelBinding | None = None (empty)

35 

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

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

38 

39 else: 

40 

41 class Config: 

42 extra = "allow" 

43 

44 @overload (empty)

45 @classmethod (empty)

46 def from_sub(cls, binding: None) -> None: ... (empty)

47 

48 @overload (empty)

49 @classmethod (empty)

50 def from_sub(cls, binding: SpecBinding) -> Self: ... (empty)

51 

52 @classmethod (empty)

53 def from_sub(cls, binding: SpecBinding | None) -> Self | None: (empty)

54 if binding is None: 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true(empty)

55 return None 

56 

57 if binding.amqp and (amqp := amqp_bindings.ChannelBinding.from_sub(binding.amqp)): (empty)

58 return cls(amqp=amqp) (empty)

59 

60 if binding.kafka and ( (empty)

61 kafka := kafka_bindings.ChannelBinding.from_sub(binding.kafka) 

62 ): 

63 return cls(kafka=kafka) (empty)

64 

65 if binding.mqtt and (mqtt := mqtt_bindings.ChannelBinding.from_sub(binding.mqtt)): (empty)

66 return cls(mqtt=mqtt) (empty)

67 

68 if binding.nats and (nats := nats_bindings.ChannelBinding.from_sub(binding.nats)): (empty)

69 return cls(nats=nats) (empty)

70 

71 if binding.redis and ( 71 ↛ 76line 71 didn't jump to line 76 because the condition on line 71 was always true(empty)

72 redis := redis_bindings.ChannelBinding.from_sub(binding.redis) 

73 ): 

74 return cls(redis=redis) (empty)

75 

76 if binding.sqs and (sqs := sqs_bindings.ChannelBinding.from_sub(binding.sqs)): 

77 return cls(sqs=sqs) 

78 

79 return None 

80 

81 @overload (empty)

82 @classmethod (empty)

83 def from_pub(cls, binding: None) -> None: ... (empty)

84 

85 @overload (empty)

86 @classmethod (empty)

87 def from_pub(cls, binding: SpecBinding) -> Self: ... (empty)

88 

89 @classmethod (empty)

90 def from_pub(cls, binding: SpecBinding | None) -> Self | None: (empty)

91 if binding is None: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true(empty)

92 return None 

93 

94 if binding.amqp and (amqp := amqp_bindings.ChannelBinding.from_pub(binding.amqp)): (empty)

95 return cls(amqp=amqp) (empty)

96 

97 if binding.kafka and ( (empty)

98 kafka := kafka_bindings.ChannelBinding.from_pub(binding.kafka) 

99 ): 

100 return cls(kafka=kafka) (empty)

101 

102 if binding.mqtt and (mqtt := mqtt_bindings.ChannelBinding.from_pub(binding.mqtt)): (empty)

103 return cls(mqtt=mqtt) (empty)

104 

105 if binding.nats and (nats := nats_bindings.ChannelBinding.from_pub(binding.nats)): (empty)

106 return cls(nats=nats) (empty)

107 

108 if binding.redis and ( 108 ↛ 113line 108 didn't jump to line 113 because the condition on line 108 was always true(empty)

109 redis := redis_bindings.ChannelBinding.from_pub(binding.redis) 

110 ): 

111 return cls(redis=redis) (empty)

112 

113 if binding.sqs and (sqs := sqs_bindings.ChannelBinding.from_pub(binding.sqs)): 

114 return cls(sqs=sqs) 

115 

116 return None