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

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 

8 

9 

10class DraftFDWarning(RuntimeWarning): 

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

12 

13 .. versionadded:: 27 

14 """ 

15 

16 def __init__(self, msg=""): 

17 if not msg: 

18 msg = ( 

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

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

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

22 ) 

23 super().__init__(msg) 

24 

25 

26class ZMQBaseError(Exception): 

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

28 

29 

30class ZMQError(ZMQBaseError): 

31 """Wrap an errno style error. 

32 

33 Parameters 

34 ---------- 

35 errno : int 

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

37 used. 

38 msg : str 

39 Description of the error or None. 

40 """ 

41 

42 errno: int | None = None 

43 strerror: str 

44 

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

46 """Wrap an errno style error. 

47 

48 Parameters 

49 ---------- 

50 errno : int 

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

52 used. 

53 msg : string 

54 Description of the error or None. 

55 """ 

56 from zmq.backend import strerror, zmq_errno 

57 

58 if errno is None: 

59 errno = zmq_errno() 

60 if isinstance(errno, int): 

61 self.errno = errno 

62 if msg is None: 

63 self.strerror = strerror(errno) 

64 else: 

65 self.strerror = msg 

66 else: 

67 if msg is None: 

68 self.strerror = str(errno) 

69 else: 

70 self.strerror = msg 

71 # flush signals, because there could be a SIGINT 

72 # waiting to pounce, resulting in uncaught exceptions. 

73 # Doing this here means getting SIGINT during a blocking 

74 # libzmq call will raise a *catchable* KeyboardInterrupt 

75 # PyErr_CheckSignals() 

76 

77 def __str__(self) -> str: 

78 return self.strerror 

79 

80 def __repr__(self) -> str: 

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

82 

83 

84class ZMQBindError(ZMQBaseError): 

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

86 

87 See Also 

88 -------- 

89 .Socket.bind_to_random_port 

90 """ 

91 

92 

93class NotDone(ZMQBaseError): 

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

95 

96 See Also 

97 -------- 

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

99 """ 

100 

101 

102class ContextTerminated(ZMQError): 

103 """Wrapper for zmq.ETERM 

104 

105 .. versionadded:: 13.0 

106 """ 

107 

108 def __init__(self, errno="ignored", msg="ignored"): 

109 from zmq import ETERM 

110 

111 super().__init__(ETERM) 

112 

113 

114class Again(ZMQError): 

115 """Wrapper for zmq.EAGAIN 

116 

117 .. versionadded:: 13.0 

118 """ 

119 

120 def __init__(self, errno="ignored", msg="ignored"): 

121 from zmq import EAGAIN 

122 

123 super().__init__(EAGAIN) 

124 

125 

126class InterruptedSystemCall(ZMQError, InterruptedError): 

127 """Wrapper for EINTR 

128 

129 This exception should be caught internally in pyzmq 

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

131 

132 .. versionadded:: 14.7 

133 """ 

134 

135 errno = EINTR 

136 strerror: str 

137 

138 def __init__(self, errno="ignored", msg="ignored"): 

139 super().__init__(EINTR) 

140 

141 def __str__(self): 

142 s = super().__str__() 

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

144 

145 

146def _check_rc(rc, errno=None, error_without_errno=True): 

147 """internal utility for checking zmq return condition 

148 

149 and raising the appropriate Exception class 

150 """ 

151 if rc == -1: 

152 if errno is None: 

153 from zmq.backend import zmq_errno 

154 

155 errno = zmq_errno() 

156 if errno == 0 and not error_without_errno: 

157 return 

158 from zmq import EAGAIN, ETERM 

159 

160 if errno == EINTR: 

161 raise InterruptedSystemCall(errno) 

162 elif errno == EAGAIN: 

163 raise Again(errno) 

164 elif errno == ETERM: 

165 raise ContextTerminated(errno) 

166 else: 

167 raise ZMQError(errno) 

168 

169 

170_zmq_version_info = None 

171_zmq_version = None 

172 

173 

174class ZMQVersionError(NotImplementedError): 

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

176 

177 .. versionadded:: 14.2 

178 """ 

179 

180 min_version = None 

181 

182 def __init__(self, min_version: str, msg: str = "Feature"): 

183 global _zmq_version 

184 if _zmq_version is None: 

185 from zmq import zmq_version 

186 

187 _zmq_version = zmq_version() 

188 self.msg = msg 

189 self.min_version = min_version 

190 self.version = _zmq_version 

191 

192 def __repr__(self): 

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

194 

195 def __str__(self): 

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

197 

198 

199def _check_version( 

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

201 msg: str = "Feature", 

202): 

203 """Check for libzmq 

204 

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

206 

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

208 """ 

209 global _zmq_version_info 

210 if _zmq_version_info is None: 

211 from zmq import zmq_version_info 

212 

213 _zmq_version_info = zmq_version_info() 

214 if _zmq_version_info < min_version_info: 

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

216 raise ZMQVersionError(min_version, msg) 

217 

218 

219__all__ = [ 

220 "DraftFDWarning", 

221 "ZMQBaseError", 

222 "ZMQBindError", 

223 "ZMQError", 

224 "NotDone", 

225 "ContextTerminated", 

226 "InterruptedSystemCall", 

227 "Again", 

228 "ZMQVersionError", 

229]