Coverage for faststream / confluent / subscriber / specification.py: 78%

1from faststream._internal.endpoint.subscriber import SubscriberSpecification (empty)

2from faststream.confluent.configs import KafkaBrokerConfig (empty)

3from faststream.specification.asyncapi.utils import resolve_payloads (empty)

4from faststream.specification.schema import Message, Operation, SubscriberSpec (empty)

5from faststream.specification.schema.bindings import ChannelBinding, kafka (empty)

6 

7from .config import KafkaSubscriberSpecificationConfig (empty)

8 

9 

10class KafkaSubscriberSpecification( (empty)

11 SubscriberSpecification[KafkaBrokerConfig, KafkaSubscriberSpecificationConfig], 

12): 

13 @property (empty)

14 def topics(self) -> list[str]: (empty)

15 topics: set[str] = set() (empty)

16 

17 topics.update(f"{self._outer_config.prefix}{t}" for t in self.config.topics) (empty)

18 

19 topics.update( (empty)

20 f"{self._outer_config.prefix}{p.topic}" for p in self.config.partitions 

21 ) 

22 

23 return list(topics) (empty)

24 

25 @property (empty)

26 def name(self) -> str: (empty)

27 if self.config.title_: 

28 return self.config.title_ 

29 

30 return f"{','.join(self.topics)}:{self.call_name}" 

31 

32 def get_schema(self) -> dict[str, SubscriberSpec]: (empty)

33 payloads = self.get_payloads() (empty)

34 

35 channels = {} (empty)

36 for t in self.topics: (empty)

37 handler_name = self.config.title_ or f"{t}:{self.call_name}" (empty)

38 

39 channels[handler_name] = SubscriberSpec( (empty)

40 description=self.description, 

41 operation=Operation( 

42 message=Message( 

43 title=f"{handler_name}:Message", 

44 payload=resolve_payloads(payloads), 

45 ), 

46 bindings=None, 

47 ), 

48 bindings=ChannelBinding( 

49 kafka=kafka.ChannelBinding( 

50 topic=t, 

51 partitions=None, 

52 replicas=None, 

53 ), 

54 ), 

55 ) 

56 

57 return channels (empty)