Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/zmq/error.py: 46%

1"""0MQ Error classes and functions.""" 

2 

3# Copyright (C) PyZMQ Developers 

4# Distributed under the terms of the Modified BSD License. 

5from __future__ import annotations 

6 

7from errno import EINTR 

8from typing import Literal 

9 

10 

11class DraftFDWarning(RuntimeWarning): 

12 """Warning for using experimental FD on draft sockets. 

13 

14 .. versionadded:: 27 

15 """ 

16 

17 def __init__(self, msg: str = "") -> None: 

18 if not msg: 

19 msg = ( 

20 "pyzmq's back-fill socket.FD support on thread-safe sockets is experimental, and may be removed." 

21 " This warning will go away automatically if/when libzmq implements socket.FD on thread-safe sockets." 

22 " You can suppress this warning with `warnings.simplefilter('ignore', zmq.error.DraftFDWarning)" 

23 ) 

24 super().__init__(msg) 

25 

26 

27class ZMQBaseError(Exception): 

28 """Base exception class for 0MQ errors in Python.""" 

29 

30 

31class ZMQError(ZMQBaseError): 

32 """Wrap an errno style error. 

33 

34 Parameters 

35 ---------- 

36 errno : int 

37 The ZMQ errno or None. If None, then ``zmq_errno()`` is called and 

38 used. 

39 msg : str 

40 Description of the error or None. 

41 """ 

42 

43 errno: int | None = None 

44 strerror: str 

45 

46 def __init__(self, errno: int | None = None, msg: str | None = None): 

47 """Wrap an errno style error. 

48 

49 Parameters 

50 ---------- 

51 errno : int 

52 The ZMQ errno or None. If None, then ``zmq_errno()`` is called and 

53 used. 

54 msg : string 

55 Description of the error or None. 

56 """ 

57 from zmq.backend import strerror, zmq_errno 

58 

59 if errno is None: 

60 errno = zmq_errno() 

61 if isinstance(errno, int): 

62 self.errno = errno 

63 if msg is None: 

64 self.strerror = strerror(errno) 

65 else: 

66 self.strerror = msg 

67 else: 

68 if msg is None: 

69 self.strerror = str(errno) 

70 else: 

71 self.strerror = msg 

72 # flush signals, because there could be a SIGINT 

73 # waiting to pounce, resulting in uncaught exceptions. 

74 # Doing this here means getting SIGINT during a blocking 

75 # libzmq call will raise a *catchable* KeyboardInterrupt 

76 # PyErr_CheckSignals() 

77 

78 def __str__(self) -> str: 

79 return self.strerror 

80 

81 def __repr__(self) -> str: 

82 return f"{self.__class__.__name__}('{str(self)}')" 

83 

84 

85class ZMQBindError(ZMQBaseError): 

86 """An error for ``Socket.bind_to_random_port()``. 

87 

88 See Also 

89 -------- 

90 .Socket.bind_to_random_port 

91 """ 

92 

93 

94class NotDone(ZMQBaseError): 

95 """Raised when timeout is reached while waiting for 0MQ to finish with a Message 

96 

97 See Also 

98 -------- 

99 .MessageTracker.wait : object for tracking when ZeroMQ is done 

100 """ 

101 

102 

103class ContextTerminated(ZMQError): 

104 """Wrapper for zmq.ETERM 

105 

106 .. versionadded:: 13.0 

107 """ 

108 

109 def __init__( 

110 self, 

111 errno: int | Literal["ignored"] = "ignored", 

112 msg: str = "ignored", 

113 ) -> None: 

114 from zmq import ETERM 

115 

116 super().__init__(ETERM) 

117 

118 

119class Again(ZMQError): 

120 """Wrapper for zmq.EAGAIN 

121 

122 .. versionadded:: 13.0 

123 """ 

124 

125 def __init__( 

126 self, 

127 errno: int | Literal["ignored"] = "ignored", 

128 msg: str = "ignored", 

129 ) -> None: 

130 from zmq import EAGAIN 

131 

132 super().__init__(EAGAIN) 

133 

134 

135class InterruptedSystemCall(ZMQError, InterruptedError): 

136 """Wrapper for EINTR 

137 

138 This exception should be caught internally in pyzmq 

139 to retry system calls, and not propagate to the user. 

140 

141 .. versionadded:: 14.7 

142 """ 

143 

144 errno = EINTR 

145 strerror: str 

146 

147 def __init__( 

148 self, 

149 errno: int | Literal["ignored"] = "ignored", 

150 msg: str = "ignored", 

151 ) -> None: 

152 super().__init__(EINTR) 

153 

154 def __str__(self) -> str: 

155 s = super().__str__() 

156 return s + ": This call should have been retried. Please report this to pyzmq." 

157 

158 

159def _check_rc(rc, errno: int | None = None, error_without_errno: bool = True) -> None: 

160 """internal utility for checking zmq return condition 

161 

162 and raising the appropriate Exception class 

163 """ 

164 if rc == -1: 

165 if errno is None: 

166 from zmq.backend import zmq_errno 

167 

168 errno = zmq_errno() 

169 if errno == 0 and not error_without_errno: 

170 return 

171 from zmq import EAGAIN, ETERM 

172 

173 if errno == EINTR: 

174 raise InterruptedSystemCall(errno) 

175 elif errno == EAGAIN: 

176 raise Again(errno) 

177 elif errno == ETERM: 

178 raise ContextTerminated(errno) 

179 else: 

180 raise ZMQError(errno) 

181 

182 

183_zmq_version_info = None 

184_zmq_version = None 

185 

186 

187class ZMQVersionError(NotImplementedError): 

188 """Raised when a feature is not provided by the linked version of libzmq. 

189 

190 .. versionadded:: 14.2 

191 """ 

192 

193 msg: str 

194 min_version: str 

195 version: str 

196 

197 def __init__(self, min_version: str, msg: str = "Feature") -> None: 

198 global _zmq_version 

199 if _zmq_version is None: 

200 from zmq import zmq_version 

201 

202 _zmq_version = zmq_version() 

203 self.msg = msg 

204 self.min_version = min_version 

205 self.version = _zmq_version 

206 

207 def __repr__(self): 

208 return f"ZMQVersionError('{str(self)}')" 

209 

210 def __str__(self): 

211 return f"{self.msg} requires libzmq >= {self.min_version}, have {self.version}" 

212 

213 

214def _check_version( 

215 min_version_info: tuple[int] | tuple[int, int] | tuple[int, int, int], 

216 msg: str = "Feature", 

217) -> None: 

218 """Check for libzmq 

219 

220 raises ZMQVersionError if current zmq version is not at least min_version 

221 

222 min_version_info is a tuple of integers, and will be compared against zmq.zmq_version_info(). 

223 """ 

224 global _zmq_version_info 

225 if _zmq_version_info is None: 

226 from zmq import zmq_version_info 

227 

228 _zmq_version_info = zmq_version_info() 

229 if _zmq_version_info < min_version_info: 

230 min_version = ".".join(str(v) for v in min_version_info) 

231 raise ZMQVersionError(min_version, msg) 

232 

233 

234__all__ = [ 

235 "DraftFDWarning", 

236 "ZMQBaseError", 

237 "ZMQBindError", 

238 "ZMQError", 

239 "NotDone", 

240 "ContextTerminated", 

241 "InterruptedSystemCall", 

242 "Again", 

243 "ZMQVersionError", 

244]