Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/util/queue.py: 51%

1# util/queue.py 

2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# 

5# This module is part of SQLAlchemy and is released under 

6# the MIT License: https://www.opensource.org/licenses/mit-license.php 

7# mypy: allow-untyped-defs, allow-untyped-calls 

8 

9"""An adaptation of Py2.3/2.4's Queue module which supports reentrant 

10behavior, using RLock instead of Lock for its mutex object. The 

11Queue object is used exclusively by the sqlalchemy.pool.QueuePool 

12class. 

13 

14This is to support the connection pool's usage of weakref callbacks to return 

15connections to the underlying Queue, which can in extremely 

16rare cases be invoked within the ``get()`` method of the Queue itself, 

17producing a ``put()`` inside the ``get()`` and therefore a reentrant 

18condition. 

19 

20""" 

21 

22from __future__ import annotations 

23 

24import asyncio 

25from collections import deque 

26import threading 

27from time import time as _time 

28from typing import Any 

29from typing import Deque 

30from typing import Generic 

31from typing import Optional 

32from typing import TypeVar 

33 

34from . import compat 

35from .concurrency import await_ 

36from .langhelpers import memoized_property 

37 

38_T = TypeVar("_T", bound=Any) 

39__all__ = ["Empty", "Full", "Queue"] 

40 

41 

42class Empty(Exception): 

43 "Exception raised by Queue.get(block=0)/get_nowait()." 

44 

45 pass 

46 

47 

48class Full(Exception): 

49 "Exception raised by Queue.put(block=0)/put_nowait()." 

50 

51 pass 

52 

53 

54class QueueCommon(Generic[_T]): 

55 maxsize: int 

56 use_lifo: bool 

57 

58 def __init__(self, maxsize: int = 0, use_lifo: bool = False): ... 

59 

60 def empty(self) -> bool: 

61 raise NotImplementedError() 

62 

63 def full(self) -> bool: 

64 raise NotImplementedError() 

65 

66 def qsize(self) -> int: 

67 raise NotImplementedError() 

68 

69 def put_nowait(self, item: _T) -> None: 

70 raise NotImplementedError() 

71 

72 def put( 

73 self, item: _T, block: bool = True, timeout: Optional[float] = None 

74 ) -> None: 

75 raise NotImplementedError() 

76 

77 def get_nowait(self) -> _T: 

78 raise NotImplementedError() 

79 

80 def get(self, block: bool = True, timeout: Optional[float] = None) -> _T: 

81 raise NotImplementedError() 

82 

83 

84class Queue(QueueCommon[_T]): 

85 queue: Deque[_T] 

86 

87 def __init__(self, maxsize: int = 0, use_lifo: bool = False): 

88 """Initialize a queue object with a given maximum size. 

89 

90 If `maxsize` is <= 0, the queue size is infinite. 

91 

92 If `use_lifo` is True, this Queue acts like a Stack (LIFO). 

93 """ 

94 

95 self._init(maxsize) 

96 # mutex must be held whenever the queue is mutating. All methods 

97 # that acquire mutex must release it before returning. mutex 

98 # is shared between the two conditions, so acquiring and 

99 # releasing the conditions also acquires and releases mutex. 

100 self.mutex = threading.RLock() 

101 # Notify not_empty whenever an item is added to the queue; a 

102 # thread waiting to get is notified then. 

103 self.not_empty = threading.Condition(self.mutex) 

104 # Notify not_full whenever an item is removed from the queue; 

105 # a thread waiting to put is notified then. 

106 self.not_full = threading.Condition(self.mutex) 

107 # If this queue uses LIFO or FIFO 

108 self.use_lifo = use_lifo 

109 

110 def qsize(self) -> int: 

111 """Return the approximate size of the queue (not reliable!).""" 

112 

113 with self.mutex: 

114 return self._qsize() 

115 

116 def empty(self) -> bool: 

117 """Return True if the queue is empty, False otherwise (not 

118 reliable!).""" 

119 

120 with self.mutex: 

121 return self._empty() 

122 

123 def full(self) -> bool: 

124 """Return True if the queue is full, False otherwise (not 

125 reliable!).""" 

126 

127 with self.mutex: 

128 return self._full() 

129 

130 def put( 

131 self, item: _T, block: bool = True, timeout: Optional[float] = None 

132 ) -> None: 

133 """Put an item into the queue. 

134 

135 If optional args `block` is True and `timeout` is None (the 

136 default), block if necessary until a free slot is 

137 available. If `timeout` is a positive number, it blocks at 

138 most `timeout` seconds and raises the ``Full`` exception if no 

139 free slot was available within that time. Otherwise (`block` 

140 is false), put an item on the queue if a free slot is 

141 immediately available, else raise the ``Full`` exception 

142 (`timeout` is ignored in that case). 

143 """ 

144 

145 with self.not_full: 

146 if not block: 

147 if self._full(): 

148 raise Full 

149 elif timeout is None: 

150 while self._full(): 

151 self.not_full.wait() 

152 else: 

153 if timeout < 0: 

154 raise ValueError("'timeout' must be a positive number") 

155 endtime = _time() + timeout 

156 while self._full(): 

157 remaining = endtime - _time() 

158 if remaining <= 0.0: 

159 raise Full 

160 self.not_full.wait(remaining) 

161 self._put(item) 

162 self.not_empty.notify() 

163 

164 def put_nowait(self, item: _T) -> None: 

165 """Put an item into the queue without blocking. 

166 

167 Only enqueue the item if a free slot is immediately available. 

168 Otherwise raise the ``Full`` exception. 

169 """ 

170 return self.put(item, False) 

171 

172 def get(self, block: bool = True, timeout: Optional[float] = None) -> _T: 

173 """Remove and return an item from the queue. 

174 

175 If optional args `block` is True and `timeout` is None (the 

176 default), block if necessary until an item is available. If 

177 `timeout` is a positive number, it blocks at most `timeout` 

178 seconds and raises the ``Empty`` exception if no item was 

179 available within that time. Otherwise (`block` is false), 

180 return an item if one is immediately available, else raise the 

181 ``Empty`` exception (`timeout` is ignored in that case). 

182 

183 """ 

184 with self.not_empty: 

185 if not block: 

186 if self._empty(): 

187 raise Empty 

188 elif timeout is None: 

189 while self._empty(): 

190 self.not_empty.wait() 

191 else: 

192 if timeout < 0: 

193 raise ValueError("'timeout' must be a positive number") 

194 endtime = _time() + timeout 

195 while self._empty(): 

196 remaining = endtime - _time() 

197 if remaining <= 0.0: 

198 raise Empty 

199 self.not_empty.wait(remaining) 

200 item = self._get() 

201 self.not_full.notify() 

202 return item 

203 

204 def get_nowait(self) -> _T: 

205 """Remove and return an item from the queue without blocking. 

206 

207 Only get an item if one is immediately available. Otherwise 

208 raise the ``Empty`` exception. 

209 """ 

210 

211 return self.get(False) 

212 

213 def _init(self, maxsize: int) -> None: 

214 self.maxsize = maxsize 

215 self.queue = deque() 

216 

217 def _qsize(self) -> int: 

218 return len(self.queue) 

219 

220 def _empty(self) -> bool: 

221 return not self.queue 

222 

223 def _full(self) -> bool: 

224 return self.maxsize > 0 and len(self.queue) == self.maxsize 

225 

226 def _put(self, item: _T) -> None: 

227 self.queue.append(item) 

228 

229 def _get(self) -> _T: 

230 if self.use_lifo: 

231 # LIFO 

232 return self.queue.pop() 

233 else: 

234 # FIFO 

235 return self.queue.popleft() 

236 

237 

238class AsyncAdaptedQueue(QueueCommon[_T]): 

239 def __init__(self, maxsize: int = 0, use_lifo: bool = False): 

240 self.use_lifo = use_lifo 

241 self.maxsize = maxsize 

242 

243 def empty(self) -> bool: 

244 return self._queue.empty() 

245 

246 def full(self): 

247 return self._queue.full() 

248 

249 def qsize(self): 

250 return self._queue.qsize() 

251 

252 @memoized_property 

253 def _queue(self) -> asyncio.Queue[_T]: 

254 # Delay creation of the queue until it is first used, to avoid 

255 # binding it to a possibly wrong event loop. 

256 # By delaying the creation of the pool we accommodate the common 

257 # usage pattern of instantiating the engine at module level, where a 

258 # different event loop is in present compared to when the application 

259 # is actually run. 

260 

261 queue: asyncio.Queue[_T] 

262 

263 if self.use_lifo: 

264 queue = asyncio.LifoQueue(maxsize=self.maxsize) 

265 else: 

266 queue = asyncio.Queue(maxsize=self.maxsize) 

267 return queue 

268 

269 def put_nowait(self, item: _T) -> None: 

270 try: 

271 self._queue.put_nowait(item) 

272 except asyncio.QueueFull as err: 

273 raise Full() from err 

274 

275 def put( 

276 self, item: _T, block: bool = True, timeout: Optional[float] = None 

277 ) -> None: 

278 if not block: 

279 return self.put_nowait(item) 

280 

281 try: 

282 if timeout is not None: 

283 await_(compat.wait_for(timeout, self._queue.put, item)) 

284 else: 

285 await_(self._queue.put(item)) 

286 except (asyncio.QueueFull, asyncio.TimeoutError) as err: 

287 raise Full() from err 

288 

289 def get_nowait(self) -> _T: 

290 try: 

291 return self._queue.get_nowait() 

292 except asyncio.QueueEmpty as err: 

293 raise Empty() from err 

294 

295 def get(self, block: bool = True, timeout: Optional[float] = None) -> _T: 

296 if not block: 

297 return self.get_nowait() 

298 

299 try: 

300 # an item that's already available is taken without awaiting 

301 return self._queue.get_nowait() 

302 except asyncio.QueueEmpty: 

303 pass 

304 

305 try: 

306 if timeout is not None: 

307 return await_(compat.wait_for(timeout, self._queue.get)) 

308 else: 

309 return await_(self._queue.get()) 

310 except (asyncio.QueueEmpty, asyncio.TimeoutError) as err: 

311 raise Empty() from err