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

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

2from dataclasses import dataclass, field (empty)

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

4 

5from typing_extensions import TypeVar as TypeVar313 (empty)

6 

7from faststream._internal.constants import EMPTY (empty)

8from faststream._internal.di import FastDependsConfig (empty)

9from faststream._internal.logger import LoggerState (empty)

10from faststream._internal.producer import ProducerProto, ProducerUnset (empty)

11 

12if TYPE_CHECKING: (empty)

13 from fast_depends.dependencies import Dependant 

14 

15 from faststream._internal.parser import CodecProto 

16 from faststream._internal.types import BrokerMiddleware, CustomCallable 

17 from faststream.middlewares import AckPolicy 

18 

19 

20@dataclass(kw_only=True) (empty)

21class BrokerConfig: (empty)

22 prefix: str = "" (empty)

23 include_in_schema: bool | None = True (empty)

24 

25 broker_middlewares: Sequence["BrokerMiddleware[Any]"] = () (empty)

26 broker_parser: Optional["CustomCallable"] = None (empty)

27 broker_decoder: Optional["CustomCallable"] = None (empty)

28 broker_codec: Optional["CodecProto"] = None (empty)

29 

30 producer: "ProducerProto[Any]" = field(default_factory=ProducerUnset) (empty)

31 logger: "LoggerState" = field(default_factory=LoggerState) (empty)

32 fd_config: "FastDependsConfig" = field(default_factory=FastDependsConfig) (empty)

33 

34 # subscriber options 

35 broker_dependencies: Iterable["Dependant"] = () (empty)

36 graceful_timeout: float | None = None (empty)

37 ack_policy: "AckPolicy" = field(default_factory=lambda: EMPTY) (empty)

38 extra_context: dict[str, Any] = field(default_factory=dict) (empty)

39 

40 def __repr__(self) -> str: (empty)

41 return f"{self.__class__.__name__}(id: {id(self)})" 

42 

43 def __bool__(self) -> bool: (empty)

44 return bool( (empty)

45 self.include_in_schema is not None 

46 or self.broker_middlewares 

47 or self.broker_dependencies 

48 or self.prefix, 

49 ) 

50 

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

52 self.broker_middlewares = (*self.broker_middlewares, middleware) (empty)

53 

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

55 self.broker_middlewares = (middleware, *self.broker_middlewares) 

56 

57 

58BrokerConfigType = TypeVar313( (empty)

59 "BrokerConfigType", 

60 bound=BrokerConfig, 

61 default=BrokerConfig, 

62) 

63 

64ConfigType = Union["ConfigComposition[Any]", "BrokerConfigType", BrokerConfig] (empty)

65 

66 

67class ConfigComposition(Generic[BrokerConfigType]): # noqa: PLR0904 (empty)

68 def __init__(self, config: BrokerConfigType) -> None: (empty)

69 self.configs: tuple[ConfigType, ...] = (config,) (empty)

70 

71 @property (empty)

72 def broker_config(self) -> "BrokerConfigType": (empty)

73 assert self.configs (empty)

74 return self.configs[0] # type: ignore[return-value] (empty)

75 

76 def __repr__(self) -> str: (empty)

77 return f"{self.__class__.__name__}({', '.join(repr(c) for c in self.configs)})" 

78 

79 def add_config(self, config: "ConfigType") -> None: (empty)

80 self.configs = (config, *self.configs) (empty)

81 

82 def reset(self) -> None: (empty)

83 self.configs = (self.configs[-1],) (empty)

84 

85 # broker priority options 

86 @property (empty)

87 def producer(self) -> "ProducerProto[Any]": (empty)

88 return self.broker_config.producer (empty)

89 

90 @property (empty)

91 def logger(self) -> "LoggerState": (empty)

92 return self.broker_config.logger (empty)

93 

94 @property (empty)

95 def fd_config(self) -> "FastDependsConfig": (empty)

96 return self.broker_config.fd_config (empty)

97 

98 @fd_config.setter (empty)

99 def fd_config(self, value: "FastDependsConfig") -> None: (empty)

100 self.broker_config.fd_config = value (empty)

101 

102 @property (empty)

103 def graceful_timeout(self) -> float | None: (empty)

104 return self.broker_config.graceful_timeout (empty)

105 

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

107 self.broker_config.add_middleware(middleware) (empty)

108 

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

110 self.broker_config.insert_middleware(middleware) 

111 

112 def __getattr__(self, name: str) -> Any: (empty)

113 return getattr(self.broker_config, name) (empty)

114 

115 # first valuable option 

116 @property (empty)

117 def broker_parser(self) -> Optional["CustomCallable"]: (empty)

118 for c in self.configs: (empty)

119 if c.broker_parser: (empty)

120 return c.broker_parser (empty)

121 return None (empty)

122 

123 @property (empty)

124 def broker_decoder(self) -> Optional["CustomCallable"]: (empty)

125 for c in self.configs: (empty)

126 if c.broker_decoder: (empty)

127 return c.broker_decoder (empty)

128 return None (empty)

129 

130 @property (empty)

131 def broker_codec(self) -> Optional["CodecProto"]: (empty)

132 for c in self.configs: (empty)

133 if c.broker_codec: (empty)

134 return c.broker_codec (empty)

135 return None (empty)

136 

137 @property (empty)

138 def ack_policy(self) -> "AckPolicy": (empty)

139 for c in reversed(self.configs): (empty)

140 ack = c.ack_policy (empty)

141 if ack is not EMPTY: (empty)

142 return ack (empty)

143 return EMPTY # type: ignore[no-any-return] (empty)

144 

145 # merged options 

146 @property (empty)

147 def extra_context(self) -> dict[str, Any]: (empty)

148 context: dict[str, Any] = {} (empty)

149 for c in self.configs: (empty)

150 context |= c.extra_context (empty)

151 return context (empty)

152 

153 @property (empty)

154 def prefix(self) -> str: (empty)

155 return "".join(c.prefix for c in self.configs) (empty)

156 

157 @property (empty)

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

159 return all(c.include_in_schema is not False for c in self.configs) (empty)

160 

161 @property (empty)

162 def broker_middlewares(self) -> Sequence["BrokerMiddleware[Any]"]: (empty)

163 return [m for c in self.configs for m in c.broker_middlewares] (empty)

164 

165 @property (empty)

166 def broker_dependencies(self) -> Iterable["Dependant"]: (empty)

167 return (b for c in self.configs for b in c.broker_dependencies) (empty)