Coverage for faststream / _internal / endpoint / subscriber / specification.py: 91%

1from abc import abstractmethod (empty)

2from typing import TYPE_CHECKING, Any, Generic (empty)

3 

4from typing_extensions import ( (empty)

5 TypeVar as TypeVar313, 

6) 

7 

8from faststream._internal.configs import BrokerConfig, SubscriberSpecificationConfig (empty)

9from faststream.exceptions import SetupError (empty)

10from faststream.specification.asyncapi.message import parse_handler_params (empty)

11from faststream.specification.asyncapi.utils import to_camelcase (empty)

12 

13if TYPE_CHECKING: (empty)

14 from faststream._internal.endpoint.subscriber.call_item import ( 

15 CallsCollection, 

16 ) 

17 from faststream.specification.schema import SubscriberSpec 

18 

19 

20T_SpecificationConfig = TypeVar313( (empty)

21 "T_SpecificationConfig", 

22 bound=SubscriberSpecificationConfig, 

23 default=SubscriberSpecificationConfig, 

24) 

25T_BrokerConfig = TypeVar313("T_BrokerConfig", bound=BrokerConfig, default=BrokerConfig) (empty)

26 

27 

28class SubscriberSpecification(Generic[T_BrokerConfig, T_SpecificationConfig]): (empty)

29 def __init__( (empty)

30 self, 

31 _outer_config: "T_BrokerConfig", 

32 specification_config: "T_SpecificationConfig", 

33 calls: "CallsCollection[Any]", 

34 ) -> None: 

35 self.calls = calls (empty)

36 self.config = specification_config (empty)

37 self._outer_config = _outer_config (empty)

38 

39 @property (empty)

40 def include_in_schema(self) -> bool: (empty)

41 return bool( (empty)

42 self._outer_config.include_in_schema and self.config.include_in_schema, 

43 ) 

44 

45 @property (empty)

46 def description(self) -> str | None: (empty)

47 return self.config.description_ or self.calls.description (empty)

48 

49 @property (empty)

50 def call_name(self) -> str: (empty)

51 return self.calls.name or "Subscriber" (empty)

52 

53 def get_payloads(self) -> list[tuple["dict[str, Any]", str]]: (empty)

54 payloads: list[tuple[dict[str, Any], str]] = [] (empty)

55 

56 call_name = self.call_name (empty)

57 

58 for h in self.calls: (empty)

59 if h.dependant is None: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true(empty)

60 msg = "You should setup `Handler` at first." 

61 raise SetupError(msg) 

62 

63 body = parse_handler_params( (empty)

64 h.dependant, 

65 prefix=f"{self.config.title_ or call_name}:Message", 

66 ) 

67 payloads.append((body, to_camelcase(h.name))) (empty)

68 

69 if not self.calls: (empty)

70 payloads.append( (empty)

71 ( 

72 { 

73 "title": f"{self.config.title_ or call_name}:Message:Payload", 

74 }, 

75 to_camelcase(call_name), 

76 ), 

77 ) 

78 

79 return payloads (empty)

80 

81 @property (empty)

82 @abstractmethod (empty)

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

84 raise NotImplementedError 

85 

86 @abstractmethod (empty)

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

88 raise NotImplementedError