Coverage for faststream / _internal / broker / registrator.py: 98%

1from abc import abstractmethod (empty)

2from collections.abc import Iterable, Sequence (empty)

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

4from weakref import WeakSet (empty)

5 

6from faststream._internal.configs import BrokerConfig, BrokerConfigType, ConfigComposition (empty)

7from faststream._internal.types import BrokerMiddleware, MsgType (empty)

8 

9if TYPE_CHECKING: (empty)

10 from fast_depends.dependencies import Dependant 

11 

12 from faststream._internal.endpoint.publisher import PublisherUsecase 

13 from faststream._internal.endpoint.subscriber import SubscriberUsecase 

14 

15 

16class Registrator(Generic[MsgType, BrokerConfigType]): (empty)

17 """Basic class for brokers and routers. 

18 

19 Contains subscribers & publishers registration logic only. 

20 """ 

21 

22 def __init__( (empty)

23 self, 

24 *, 

25 config: BrokerConfigType, 

26 routers: Iterable["Registrator[MsgType]"], 

27 ) -> None: 

28 self._parser = config.broker_parser (empty)

29 self._decoder = config.broker_decoder (empty)

30 

31 self.config: ConfigComposition[BrokerConfigType] = ConfigComposition(config) (empty)

32 

33 self._subscribers: WeakSet[SubscriberUsecase[MsgType]] = WeakSet() (empty)

34 self._publishers: WeakSet[PublisherUsecase] = WeakSet() (empty)

35 self.routers: list[Registrator[MsgType, Any]] = [] (empty)

36 

37 self.__persistent_subscribers: list[SubscriberUsecase[MsgType]] = [] (empty)

38 self.__persistent_publishers: list[PublisherUsecase] = [] (empty)

39 

40 self.__parent: Registrator[MsgType, Any] | None = None (empty)

41 

42 self.include_routers(*routers) (empty)

43 

44 @property (empty)

45 def subscribers(self) -> list["SubscriberUsecase[MsgType]"]: (empty)

46 return [*self._subscribers, *(sub for r in self.routers for sub in r.subscribers)] (empty)

47 

48 @property (empty)

49 def publishers(self) -> list["PublisherUsecase"]: (empty)

50 return [*self._publishers, *(pub for r in self.routers for pub in r.publishers)] (empty)

51 

52 def add_middleware(self, middleware: "BrokerMiddleware[Any, Any]") -> None: (empty)

53 """Append BrokerMiddleware to the end of middlewares list. 

54 

55 Current middleware will be used as a most inner of the stack. 

56 """ 

57 self.config.add_middleware(middleware) (empty)

58 

59 def insert_middleware(self, middleware: "BrokerMiddleware[Any, Any]") -> None: (empty)

60 """Insert BrokerMiddleware to the start of middlewares list. 

61 

62 Current middleware will be used as a most outer of the stack. 

63 """ 

64 self.config.insert_middleware(middleware) 

65 

66 @abstractmethod (empty)

67 def subscriber( (empty)

68 self, 

69 subscriber: "SubscriberUsecase[MsgType]", (empty)

70 persistent: bool = True, (empty)

71 ) -> "SubscriberUsecase[MsgType]": (empty)

72 self._subscribers.add(subscriber) (empty)

73 if persistent: (empty)

74 self.__persistent_subscribers.append(subscriber) (empty)

75 return subscriber (empty)

76 

77 @abstractmethod (empty)

78 def publisher( (empty)

79 self, 

80 publisher: "PublisherUsecase", (empty)

81 persistent: bool = True, (empty)

82 ) -> "PublisherUsecase": (empty)

83 self._publishers.add(publisher) (empty)

84 if persistent: (empty)

85 self.__persistent_publishers.append(publisher) (empty)

86 return publisher (empty)

87 

88 def include_router( (empty)

89 self, 

90 router: "Registrator[MsgType, Any]", 

91 *, 

92 prefix: str = "", 

93 dependencies: Iterable["Dependant"] = (), 

94 middlewares: Sequence["BrokerMiddleware[MsgType]"] = (), 

95 include_in_schema: bool | None = None, 

96 ) -> None: 

97 """Includes a router in the current object.""" 

98 if router.parent is self: (empty)

99 return (empty)

100 router.parent = self (empty)

101 

102 if options_config := BrokerConfig( (empty)

103 prefix=prefix, 

104 include_in_schema=include_in_schema, 

105 broker_middlewares=middlewares, 

106 broker_dependencies=dependencies, 

107 ): 

108 router.config.add_config(options_config) (empty)

109 

110 router.config.add_config(self.config) (empty)

111 self.routers.append(router) (empty)

112 

113 @property (empty)

114 def parent(self) -> "Registrator[MsgType, Any] | None": (empty)

115 return self.__parent (empty)

116 

117 @parent.setter (empty)

118 def parent(self, parent: "Registrator[MsgType, Any]") -> None: (empty)

119 if self.__parent is not None and parent is not self.__parent: (empty)

120 self.__parent.routers.remove(self) (empty)

121 self.config.reset() (empty)

122 self.__parent = parent (empty)

123 

124 def include_routers( (empty)

125 self, 

126 *routers: "Registrator[MsgType, Any]", 

127 ) -> None: 

128 """Includes routers in the object.""" 

129 for r in routers: (empty)

130 self.include_router(r) (empty)