Coverage for faststream / specification / asyncapi / v2_6_0 / schema / bindings / redis / channel.py: 80%

1"""AsyncAPI Redis bindings. 

2 

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

4""" 

5 

6from pydantic import BaseModel (empty)

7from typing_extensions import Self (empty)

8 

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

10 

11 

12class ChannelBinding(BaseModel): (empty)

13 """A class to represent channel binding. 

14 

15 Attributes: 

16 channel : the channel name 

17 method : the method used for binding (ssubscribe, psubscribe, subscribe) 

18 bindingVersion : the version of the binding 

19 """ 

20 

21 channel: str (empty)

22 method: str | None = None (empty)

23 groupName: str | None = None (empty)

24 consumerName: str | None = None (empty)

25 bindingVersion: str = "custom" (empty)

26 

27 @classmethod (empty)

28 def from_sub(cls, binding: redis.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 channel=binding.channel, 

34 method=binding.method, 

35 groupName=binding.group_name, 

36 consumerName=binding.consumer_name, 

37 ) 

38 

39 @classmethod (empty)

40 def from_pub(cls, binding: redis.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 channel=binding.channel, 

46 method=binding.method, 

47 groupName=binding.group_name, 

48 consumerName=binding.consumer_name, 

49 )