Coverage for faststream / exceptions.py: 91%

1from collections.abc import Iterable (empty)

2from pprint import pformat (empty)

3from typing import Any (empty)

4 

5 

6class FastStreamException(Exception): # noqa: N818 (empty)

7 """Basic FastStream exception class.""" 

8 

9 

10class IgnoredException(FastStreamException): (empty)

11 """Basic Exception class ignoring by watcher context and log middleware.""" 

12 

13 

14class StopConsume(IgnoredException): (empty)

15 """Raise it to stop Handler consuming.""" 

16 

17 def __str__(self) -> str: (empty)

18 return "Consumer was stopped" (empty)

19 

20 

21class StopApplication(IgnoredException, SystemExit): (empty)

22 """Raise it to stop FastStream application.""" 

23 

24 def __str__(self) -> str: (empty)

25 return "Application was stopped" 

26 

27 

28class HandlerException(IgnoredException): (empty)

29 """Base Handler Exception.""" 

30 

31 

32class SkipMessage(HandlerException): (empty)

33 """Watcher Instruction to skip message.""" 

34 

35 def __str__(self) -> str: (empty)

36 return "Message was skipped" (empty)

37 

38 

39class AckMessage(HandlerException): (empty)

40 """Exception raised to acknowledge a message immediately. (empty)

41 

42 This exception can be used to ack a message with additional options. 

43 To watch all allowed parameters, please take a look at your broker `message.ack(**extra_options)` method 

44 signature. 

45 

46 Args: 

47 extra_options (Any): Additional parameters that will be passed to `message.ack(**extra_options)` method. 

48 """ 

49 

50 def __init__(self, **extra_options: Any) -> None: (empty)

51 self.extra_options = extra_options (empty)

52 super().__init__() (empty)

53 

54 def __str__(self) -> str: (empty)

55 return "Message was acked" (empty)

56 

57 

58class NackMessage(HandlerException): (empty)

59 """Exception raised to negatively acknowledge a message immediately. (empty)

60 

61 This exception can be used to nack a message with additional options. 

62 To watch all allowed parameters, please take a look to your broker's `message.nack(**extra_options)` method 

63 signature. 

64 

65 Args: 

66 kwargs (Any): Additional parameters that will be passed to `message.nack(**extra_options)` method. 

67 """ 

68 

69 def __init__(self, **kwargs: Any) -> None: (empty)

70 self.extra_options = kwargs (empty)

71 super().__init__() (empty)

72 

73 def __str__(self) -> str: (empty)

74 return "Message was nacked" (empty)

75 

76 

77class RejectMessage(HandlerException): (empty)

78 """Exception raised to reject a message immediately. (empty)

79 

80 This exception can be used to reject a message with additional options. 

81 To watch all allowed parameters, please take a look to your broker's `message.reject(**extra_options)` method 

82 signature. 

83 

84 Args: 

85 kwargs (Any): Additional parameters that will be passed to `message.reject(**extra_options)` method. 

86 """ 

87 

88 def __init__(self, **kwargs: Any) -> None: (empty)

89 self.extra_options = kwargs (empty)

90 super().__init__() (empty)

91 

92 def __str__(self) -> str: (empty)

93 return "Message was rejected" (empty)

94 

95 

96class SetupError(FastStreamException, ValueError): (empty)

97 """Exception to raise at wrong method usage.""" 

98 

99 

100class StartupValidationError(FastStreamException, ValueError): (empty)

101 """Exception to raise at startup hook validation error.""" 

102 

103 def __init__( (empty)

104 self, 

105 missed_fields: Iterable[str] = (), 

106 invalid_fields: Iterable[str] = (), 

107 ) -> None: 

108 self.missed_fields = missed_fields 

109 self.invalid_fields = invalid_fields 

110 

111 def __str__(self) -> str: (empty)

112 return ( 

113 f"\n Incorrect options `{' / '.join(f'--{i}' for i in (*self.missed_fields, *self.invalid_fields))}`" 

114 "\n You registered extra options in your application `lifespan/on_startup` hook, but set them wrong in CLI." 

115 ) 

116 

117 

118class FeatureNotSupportedException(FastStreamException, NotImplementedError): # noqa: N818 (empty)

119 """Raises at planned NotImplemented operation call.""" 

120 

121 

122class SubscriberNotFound(FastStreamException): (empty)

123 """Raises as a service message or in tests.""" 

124 

125 

126class IncorrectState(FastStreamException): (empty)

127 """Raises in FSM at wrong state calling.""" 

128 

129 

130class ContextError(FastStreamException, KeyError): (empty)

131 """Raises if context exception occurred.""" 

132 

133 def __init__(self, context: Any, field: str) -> None: (empty)

134 self.context = context (empty)

135 self.field = field (empty)

136 

137 def __str__(self) -> str: (empty)

138 return "".join( 

139 ( 

140 f"\n Key `{self.field}` not found in the context\n ", 

141 pformat(self.context), 

142 ), 

143 ) 

144 

145 

146WRONG_PUBLISH_ARGS = SetupError( (empty)

147 "You should use `reply_to` to send response to long-living queue " 

148 "and `rpc` to get response in sync mode.", 

149) 

150 

151 

152NOT_CONNECTED_YET = "Please, `connect()` the broker first." (empty)

153 

154 

155INSTALL_YAML = """ (empty)

156To use feature which need yaml, please install dependencies:\n 

157pip install PyYAML 

158""" 

159 

160INSTALL_TOML = """ (empty)

161To use feature which need toml, please install dependencies:\n 

162pip install tomli 

163""" 

164 

165INSTALL_WATCHFILES = """ (empty)

166To use restart feature, please install dependencies:\n 

167pip install watchfiles 

168""" 

169 

170SCHEMA_NOT_SUPPORTED = "`{schema_filename}` not supported. Make sure that your schema is valid and schema version supported by FastStream" (empty)

171 

172INSTALL_FASTSTREAM_RABBIT = """ (empty)

173To use RabbitMQ with FastStream, please install dependencies:\n 

174pip install "faststream[rabbit]" 

175""" 

176 

177INSTALL_FASTSTREAM_KAFKA = """ (empty)

178To use Apache Kafka with FastStream, please install dependencies:\n 

179pip install "faststream[kafka]" 

180""" 

181 

182INSTALL_FASTSTREAM_CONFLUENT = """ (empty)

183To use Confluent Kafka with FastStream, please install dependencies:\n 

184pip install "faststream[confluent]" 

185""" 

186 

187 

188INSTALL_FASTSTREAM_REDIS = """ (empty)

189To use Redis with FastStream, please install dependencies:\n 

190pip install "faststream[redis]" 

191""" 

192 

193INSTALL_FASTSTREAM_NATS = """ (empty)

194To use NATS with FastStream, please install dependencies:\n 

195pip install "faststream[nats]" 

196""" 

197 

198INSTALL_UVICORN = """ (empty)

199To run FastStream ASGI App via CLI, please install uvicorn:\n 

200pip install uvicorn 

201"""