Coverage for faststream / specification / asyncapi / v2_6_0 / schema / bindings / mqtt / channel.py: 79%

1"""AsyncAPI MQTT bindings. 

2 

3References: https://github.com/asyncapi/bindings/tree/master/mqtt 

4""" 

5 

6from pydantic import BaseModel (empty)

7from typing_extensions import Self (empty)

8 

9from faststream.specification.schema.bindings import mqtt (empty)

10 

11 

12class ChannelBinding(BaseModel): (empty)

13 """MQTT channel binding (AsyncAPI 2.6.0). 

14 

15 Attributes: 

16 topic: The MQTT topic name. 

17 qos: QoS level for the channel (0, 1, or 2). 

18 retain: Whether messages are retained by the broker. 

19 bindingVersion: Version of the binding spec. 

20 """ 

21 

22 topic: str (empty)

23 qos: int = 0 (empty)

24 retain: bool = False (empty)

25 bindingVersion: str = "0.2.0" (empty)

26 

27 @classmethod (empty)

28 def from_sub(cls, binding: mqtt.ChannelBinding | None) -> Self | None: (empty)

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

30 return None 

31 

32 return cls( (empty)

33 topic=binding.topic, 

34 qos=binding.qos, 

35 retain=binding.retain, 

36 bindingVersion=binding.bindingVersion, 

37 ) 

38 

39 @classmethod (empty)

40 def from_pub(cls, binding: mqtt.ChannelBinding | None) -> Self | None: (empty)

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

42 return None 

43 

44 return cls( (empty)

45 topic=binding.topic, 

46 qos=binding.qos, 

47 retain=binding.retain, 

48 bindingVersion=binding.bindingVersion, 

49 )