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

1"""AsyncAPI Kafka bindings. 

2 

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

4""" 

5 

6from pydantic import BaseModel, PositiveInt (empty)

7from typing_extensions import Self (empty)

8 

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

10 

11 

12class ChannelBinding(BaseModel): (empty)

13 """A class to represent a channel binding. 

14 

15 Attributes: 

16 topic : optional string representing the topic 

17 partitions : optional positive integer representing the number of partitions 

18 replicas : optional positive integer representing the number of replicas 

19 bindingVersion : string representing the binding version 

20 """ 

21 

22 topic: str | None = None (empty)

23 partitions: PositiveInt | None = None (empty)

24 replicas: PositiveInt | None = None (empty)

25 bindingVersion: str = "0.4.0" (empty)

26 

27 # TODO: 

28 # topicConfiguration 

29 

30 @classmethod (empty)

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

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

33 return None 

34 

35 return cls( (empty)

36 topic=binding.topic, 

37 partitions=binding.partitions, 

38 replicas=binding.replicas, 

39 ) 

40 

41 @classmethod (empty)

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

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

44 return None 

45 

46 return cls( (empty)

47 topic=binding.topic, 

48 partitions=binding.partitions, 

49 replicas=binding.replicas, 

50 )