Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_soft_rw/_async.py: 39%

1"""Async wrapper around :class:`SoftReadWriteLock` for use with ``asyncio``.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import functools 

7import os 

8from contextlib import asynccontextmanager 

9from typing import TYPE_CHECKING, ParamSpec, TypeVar 

10 

11from filelock._async import ( 

12 _BackendOutcome, 

13 _capture_call, 

14 _drain_future, 

15 _future_result, 

16 _raise_cancelled_error, 

17 _wait_until_done, 

18) 

19 

20from ._sync import SoftReadWriteLock 

21 

22if TYPE_CHECKING: 

23 from collections.abc import AsyncGenerator, Callable 

24 from concurrent import futures 

25 from types import TracebackType 

26 

27 from filelock._api import AcquireReturnProxy 

28 

29_P = ParamSpec("_P") 

30_R = TypeVar("_R") 

31 

32 

33class AsyncSoftReadWriteLock: 

34 """ 

35 Async wrapper around :class:`SoftReadWriteLock` for ``asyncio`` applications. 

36 

37 The sync class's blocking filesystem operations run on a thread pool via ``loop.run_in_executor()``. The 

38 underlying :class:`SoftReadWriteLock` handles reentrancy, upgrade/downgrade rules, fork handling, heartbeat and 

39 TTL stale detection, and singleton behavior. 

40 

41 :param lock_file: path to the lock file; sidecar state/write/readers live next to it 

42 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely 

43 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately on contention 

44 :param is_singleton: if ``True``, reuse existing :class:`SoftReadWriteLock` instances per resolved path 

45 :param heartbeat_interval: seconds between heartbeat refreshes; default 30 s 

46 :param stale_threshold: seconds of mtime inactivity before a marker is stale; defaults to ``3 * heartbeat_interval`` 

47 :param poll_interval: seconds between acquire retries under contention; default 0.25 s 

48 :param loop: event loop for ``run_in_executor``; ``None`` uses the running loop 

49 :param executor: executor for ``run_in_executor``; ``None`` uses the default executor 

50 

51 .. versionadded:: 3.27.0 

52 

53 """ 

54 

55 def __init__( # ruff:ignore[too-many-arguments] # public constructor: one parameter per documented lock option 

56 self, 

57 lock_file: str | os.PathLike[str], 

58 timeout: float = -1, 

59 *, 

60 blocking: bool = True, 

61 is_singleton: bool = True, 

62 heartbeat_interval: float = 30.0, 

63 stale_threshold: float | None = None, 

64 poll_interval: float = 0.25, 

65 loop: asyncio.AbstractEventLoop | None = None, 

66 executor: futures.Executor | None = None, 

67 ) -> None: 

68 self._creator_pid = os.getpid() 

69 self._lock = SoftReadWriteLock( 

70 lock_file, 

71 timeout, 

72 blocking=blocking, 

73 is_singleton=is_singleton, 

74 heartbeat_interval=heartbeat_interval, 

75 stale_threshold=stale_threshold, 

76 poll_interval=poll_interval, 

77 ) 

78 self._loop = loop 

79 self._executor = executor 

80 

81 @property 

82 def lock_file(self) -> str: 

83 """The path to the lock file passed to the constructor.""" 

84 return self._lock.lock_file 

85 

86 @property 

87 def timeout(self) -> float: 

88 """The default timeout applied when ``acquire_read`` / ``acquire_write`` is called without one.""" 

89 return self._lock.timeout 

90 

91 @property 

92 def blocking(self) -> bool: 

93 """Whether ``acquire_*`` defaults to blocking; ``False`` makes contention raise immediately.""" 

94 return self._lock.blocking 

95 

96 @property 

97 def loop(self) -> asyncio.AbstractEventLoop | None: 

98 """The event loop used for ``run_in_executor``, or ``None`` for the running loop.""" 

99 return self._loop 

100 

101 @property 

102 def executor(self) -> futures.Executor | None: 

103 """The executor used for ``run_in_executor``, or ``None`` for the default executor.""" 

104 return self._executor 

105 

106 @asynccontextmanager 

107 async def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> AsyncGenerator[None]: 

108 """ 

109 Async context manager that acquires and releases a shared read lock. 

110 

111 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default 

112 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default 

113 

114 :raises RuntimeError: if a write lock is already held on this instance 

115 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

116 

117 """ 

118 await self.acquire_read(timeout, blocking=blocking) 

119 try: 

120 yield 

121 finally: 

122 await self.release() 

123 

124 @asynccontextmanager 

125 async def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> AsyncGenerator[None]: 

126 """ 

127 Async context manager that acquires and releases an exclusive write lock. 

128 

129 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default 

130 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default 

131 

132 :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread 

133 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

134 

135 """ 

136 await self.acquire_write(timeout, blocking=blocking) 

137 try: 

138 yield 

139 finally: 

140 await self.release() 

141 

142 async def acquire_read( 

143 self, timeout: float | None = None, *, blocking: bool | None = None 

144 ) -> AsyncAcquireSoftReadWriteReturnProxy: 

145 """ 

146 Acquire a shared read lock. 

147 

148 See :meth:`SoftReadWriteLock.acquire_read` for reentrancy / upgrade / fork semantics. The blocking work runs 

149 inside ``run_in_executor`` so other coroutines on the same loop keep progressing while this call waits. 

150 

151 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default 

152 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default 

153 

154 :returns: a proxy usable as an async context manager to release the lock 

155 

156 :raises RuntimeError: if a write lock is already held, if this instance was invalidated by 

157 :func:`os.fork`, or if :meth:`close` was called 

158 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

159 

160 """ 

161 self._raise_if_inherited() 

162 await self._run_acquire(functools.partial(self._lock.acquire_read, timeout, blocking=blocking)) 

163 return AsyncAcquireSoftReadWriteReturnProxy(lock=self) 

164 

165 async def acquire_write( 

166 self, timeout: float | None = None, *, blocking: bool | None = None 

167 ) -> AsyncAcquireSoftReadWriteReturnProxy: 

168 """ 

169 Acquire an exclusive write lock. 

170 

171 See :meth:`SoftReadWriteLock.acquire_write` for the two-phase writer-preferring semantics. The blocking work 

172 runs inside ``run_in_executor``. 

173 

174 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default 

175 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default 

176 

177 :returns: a proxy usable as an async context manager to release the lock 

178 

179 :raises RuntimeError: if a read lock is already held, if a write lock is held by a different thread, if 

180 this instance was invalidated by :func:`os.fork`, or if :meth:`close` was called 

181 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

182 

183 """ 

184 self._raise_if_inherited() 

185 await self._run_acquire(functools.partial(self._lock.acquire_write, timeout, blocking=blocking)) 

186 return AsyncAcquireSoftReadWriteReturnProxy(lock=self) 

187 

188 async def release(self, *, force: bool = False) -> None: 

189 """ 

190 Release one level of the current lock. 

191 

192 :param force: if ``True``, release the lock completely regardless of the current lock level 

193 

194 :raises RuntimeError: if no lock is currently held and *force* is ``False`` 

195 

196 """ 

197 if self._creator_pid == os.getpid(): 

198 await self._run(self._lock.release, force=force) 

199 

200 async def close(self) -> None: 

201 """Release any held lock and release the underlying filesystem resources. Idempotent.""" 

202 if self._creator_pid == os.getpid(): 

203 await self._run(self._lock.close) 

204 

205 def _raise_if_inherited(self) -> None: 

206 if self._creator_pid != os.getpid(): # pragma: forked child 

207 msg = f"AsyncSoftReadWriteLock on {self.lock_file} was inherited across fork; construct a new instance" 

208 raise RuntimeError(msg) 

209 

210 async def _run_acquire(self, acquire: Callable[[], AcquireReturnProxy]) -> None: 

211 # run_in_executor cannot recall work the pool already started, so canceling the caller does not stop the sync 

212 # acquire: it still creates its marker, sets the hold, and starts the heartbeat, which keeps the marker fresh 

213 # forever so no peer on any host can evict it as stale. Wait the submitted call out and hand the claim back, 

214 # the way AsyncReadWriteLock does. 

215 acquire_future = self._submit(acquire) 

216 try: 

217 await _wait_until_done(acquire_future) 

218 except asyncio.CancelledError as cancellation: 

219 try: 

220 await _drain_future(acquire_future) 

221 except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below 

222 _raise_cancelled_error(cancellation, error) 

223 try: 

224 await _drain_future(self._submit(self._lock.release)) 

225 except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below 

226 _raise_cancelled_error(cancellation, error) 

227 raise 

228 _future_result(acquire_future) 

229 

230 async def _run(self, func: Callable[_P, _R], *args: _P.args, **kwargs: _P.kwargs) -> _R: 

231 # A canceled release or close is already running on the pool thread; drain it so its outcome is observed 

232 # instead of finishing unwatched, then let the cancellation through. 

233 future = self._submit(func, *args, **kwargs) 

234 try: 

235 await _wait_until_done(future) 

236 except asyncio.CancelledError as cancellation: 

237 try: 

238 await _drain_future(future) 

239 except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below 

240 _raise_cancelled_error(cancellation, error) 

241 raise 

242 return _future_result(future) 

243 

244 def _submit( 

245 self, func: Callable[_P, _R], *args: _P.args, **kwargs: _P.kwargs 

246 ) -> asyncio.Future[_BackendOutcome[_R]]: 

247 loop = self._loop or asyncio.get_running_loop() 

248 return loop.run_in_executor(self._executor, _capture_call, functools.partial(func, *args, **kwargs)) 

249 

250 

251class AsyncAcquireSoftReadWriteReturnProxy: 

252 """Async context-aware object that releases an :class:`AsyncSoftReadWriteLock` on exit.""" 

253 

254 def __init__(self, lock: AsyncSoftReadWriteLock) -> None: 

255 self.lock = lock 

256 

257 async def __aenter__(self) -> AsyncSoftReadWriteLock: 

258 return self.lock 

259 

260 async def __aexit__( 

261 self, 

262 exc_type: type[BaseException] | None, 

263 exc_value: BaseException | None, 

264 traceback: TracebackType | None, 

265 ) -> None: 

266 await self.lock.release() 

267 

268 

269__all__ = [ 

270 "AsyncAcquireSoftReadWriteReturnProxy", 

271 "AsyncSoftReadWriteLock", 

272]