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

1"""Separate caller cancellation from backend task and executor-future results.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import contextlib 

7import time 

8from concurrent.futures import Future as ConcurrentFuture 

9from dataclasses import dataclass 

10from threading import Lock 

11from typing import TYPE_CHECKING, Final, Generic, NoReturn, TypeVar, cast 

12 

13from ._api import _append_exception_context, _raise_chained_errors 

14 

15if TYPE_CHECKING: 

16 from collections.abc import AsyncIterator, Awaitable, Callable 

17 

18_T = TypeVar("_T") 

19 

20 

21class _AsyncTransitionUnavailableError(Exception): 

22 pass 

23 

24 

25@dataclass(frozen=True) 

26class _BackendOutcome(Generic[_T]): 

27 value: _T | None = None 

28 error: BaseException | None = None 

29 

30 

31class _AsyncTransitionGate: 

32 def __init__(self) -> None: 

33 self._tail_lock: Final[Lock] = Lock() 

34 self._tail: ConcurrentFuture[None] | None = None 

35 

36 @contextlib.asynccontextmanager 

37 async def hold(self) -> AsyncIterator[None]: 

38 ticket: ConcurrentFuture[None] = ConcurrentFuture() 

39 with self._tail_lock: 

40 predecessor = self._tail 

41 self._tail = ticket 

42 if predecessor is not None: 

43 try: 

44 await _wait_until_done(asyncio.wrap_future(predecessor)) 

45 except asyncio.CancelledError: 

46 predecessor.add_done_callback(lambda _predecessor: self._leave(ticket)) 

47 raise 

48 try: 

49 yield 

50 finally: 

51 self._leave(ticket) 

52 

53 @contextlib.asynccontextmanager 

54 async def hold_for_acquire( 

55 self, 

56 *, 

57 blocking: bool, 

58 cancel_check: Callable[[], bool] | None, 

59 deadline: float | None, 

60 poll_interval: float, 

61 ) -> AsyncIterator[None]: 

62 ticket: ConcurrentFuture[None] = ConcurrentFuture() 

63 with self._tail_lock: 

64 predecessor = self._tail 

65 self._tail = ticket 

66 if predecessor is not None and not predecessor.done(): 

67 try: 

68 await self._wait_for_predecessor( 

69 predecessor, 

70 blocking=blocking, 

71 cancel_check=cancel_check, 

72 deadline=deadline, 

73 poll_interval=poll_interval, 

74 ) 

75 except BaseException: 

76 predecessor.add_done_callback(lambda _predecessor: self._leave(ticket)) 

77 raise 

78 try: 

79 yield 

80 finally: 

81 self._leave(ticket) 

82 

83 @staticmethod 

84 async def _wait_for_predecessor( 

85 predecessor: ConcurrentFuture[None], 

86 *, 

87 blocking: bool, 

88 cancel_check: Callable[[], bool] | None, 

89 deadline: float | None, 

90 poll_interval: float, 

91 ) -> None: 

92 if not blocking: 

93 raise _AsyncTransitionUnavailableError 

94 waiter = asyncio.wrap_future(predecessor) 

95 while not predecessor.done(): 

96 if cancel_check is not None and cancel_check(): 

97 raise _AsyncTransitionUnavailableError 

98 if deadline is not None: 

99 if (remaining := deadline - time.perf_counter()) <= 0: 

100 raise _AsyncTransitionUnavailableError 

101 wait_interval = min(poll_interval, remaining) if cancel_check is not None else remaining 

102 else: 

103 wait_interval = poll_interval if cancel_check is not None else None 

104 await asyncio.wait((waiter,), timeout=wait_interval) 

105 

106 def _leave(self, ticket: ConcurrentFuture[None]) -> None: 

107 with self._tail_lock: 

108 if self._tail is ticket: 

109 self._tail = None 

110 ticket.set_result(None) 

111 

112 

113async def _drain_future(future: asyncio.Future[_BackendOutcome[_T]]) -> _T: 

114 while not future.done(): 

115 with contextlib.suppress(asyncio.CancelledError): 

116 await _wait_until_done(future) 

117 return _future_result(future) 

118 

119 

120async def _wait_until_done(future: asyncio.Future[_T]) -> None: 

121 if not future.done(): 

122 await asyncio.wait((future,)) 

123 

124 

125def _future_result(future: asyncio.Future[_BackendOutcome[_T]]) -> _T: 

126 outcome = future.result() 

127 if (error := outcome.error) is None: 

128 return cast("_T", outcome.value) 

129 context = error.__context__ 

130 try: 

131 raise error # ruff:ignore[raise-within-try] # the handler restores context changed across the async boundary 

132 except BaseException: 

133 error.__context__ = context 

134 raise 

135 

136 

137def _capture_call(func: Callable[[], _T]) -> _BackendOutcome[_T]: 

138 try: 

139 return _BackendOutcome(value=func()) 

140 except BaseException as error: # ruff:ignore[blind-except] # backend control-flow exceptions are operation results 

141 return _BackendOutcome(error=error) 

142 

143 

144def _raise_cancelled_error(cancellation: asyncio.CancelledError, error: BaseException) -> NoReturn: 

145 # A reconciliation step failed while unwinding a cancellation, so keep both exception chains. Splice the error's 

146 # existing context onto the cancellation, then make the cancellation the error's context, so both the failure and 

147 # the cancellation that triggered it survive. Shared by the async wrappers so cancellations report the same way. 

148 if (context := error.__context__) is not None and context is not cancellation: 

149 if (cancellation_context := cancellation.__context__) is not None: 

150 _append_exception_context(context, cancellation_context) 

151 cancellation.__context__ = context 

152 error.__context__ = cancellation 

153 _raise_chained_errors(error) 

154 

155 

156async def _capture_awaitable(awaitable: Awaitable[_T]) -> _BackendOutcome[_T]: 

157 try: 

158 return _BackendOutcome(value=await awaitable) 

159 except BaseException as error: # ruff:ignore[blind-except] # backend cancellation must remain distinct from caller cancellation 

160 return _BackendOutcome(error=error) 

161 

162 

163__all__ = [ 

164 "_AsyncTransitionGate", 

165 "_AsyncTransitionUnavailableError", 

166 "_BackendOutcome", 

167 "_capture_awaitable", 

168 "_capture_call", 

169 "_drain_future", 

170 "_future_result", 

171 "_raise_cancelled_error", 

172 "_wait_until_done", 

173]