Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/asyncio.py: 25%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""An asyncio-based implementation of the file lock."""
3from __future__ import annotations
5import asyncio
6import contextlib
7import logging
8import os
9import time
10from dataclasses import dataclass
11from inspect import iscoroutinefunction
12from threading import local
13from typing import TYPE_CHECKING, Final, NoReturn, TypeVar, cast
15from ._api import (
16 _UNSET_FILE_MODE,
17 BaseFileLock,
18 CloseErrorPolicy,
19 ContextErrorPolicy,
20 FileLockContext,
21 FileLockMeta,
22 _append_exception_context,
23 _canonical,
24 _ExtraValue,
25 _fork_transition,
26 _grouped_errors,
27 _raise_body_and_release,
28 _raise_chained_errors,
29 _raise_cleanup_errors,
30 _raise_grouped_errors,
31 _register_fork_object,
32)
33from ._async import (
34 _AsyncTransitionGate,
35 _AsyncTransitionUnavailableError,
36 _BackendOutcome,
37 _capture_awaitable,
38 _capture_call,
39 _drain_future,
40 _future_result,
41 _wait_until_done,
42)
43from ._error import Timeout
44from ._lease import SoftFileLease
45from ._soft import SoftFileLock
46from ._strict import StrictSoftFileLock
47from ._unix import UnixFileLock
48from ._windows import WindowsFileLock
50if TYPE_CHECKING:
51 import sys
52 from collections.abc import Awaitable, Callable, Coroutine, Hashable
53 from concurrent import futures
54 from types import TracebackType
56 if sys.version_info >= (3, 11): # pragma: no cover (py311+)
57 from typing import Self
58 else: # pragma: no cover (<py311)
59 from typing_extensions import Self
62_LOGGER: Final[logging.Logger] = logging.getLogger("filelock")
63_ASYNC_RELEASE_CANCELLATION_ERRORS: Final[str] = "lock release cancellation and backend release both failed"
64_ASYNC_CONTEXT_RELEASE_ERRORS: Final[str] = "context body, release cancellation, and backend release failed"
65_ASYNC_RELEASE_CANCELLATION_MARKER_ATTR: Final[str] = "_filelock_async_release_cancellation"
66_ASYNC_RELEASE_CANCELLATION_MARKER: Final[list[None]] = []
68_AT = TypeVar("_AT", bound="BaseAsyncFileLock")
71class AsyncFileLockMeta(FileLockMeta):
72 def __call__( # ruff:ignore[too-many-arguments] # forwards the public constructor's documented parameters
73 cls: type[_AT], # ruff:ignore[invalid-first-argument-name-for-method] # metaclass __call__ receives the class being constructed
74 lock_file: str | os.PathLike[str],
75 timeout: float = -1,
76 mode: int = _UNSET_FILE_MODE,
77 thread_local: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument] # public API: positional bool kept for backwards compatibility
78 *,
79 blocking: bool = True,
80 is_singleton: bool = False,
81 poll_interval: float = 0.05,
82 lifetime: float | None = None,
83 context_error_policy: ContextErrorPolicy = "chain",
84 close_error_policy: CloseErrorPolicy = "default",
85 fallback_to_soft: bool = True,
86 preserve_lock_file: bool = False,
87 on_acquired: Callable[[int], None] | None = None,
88 loop: asyncio.AbstractEventLoop | None = None,
89 run_in_executor: bool = True,
90 executor: futures.Executor | None = None,
91 **kwargs: _ExtraValue,
92 ) -> _AT:
93 if thread_local and run_in_executor:
94 msg = "run_in_executor is not supported when thread_local is True"
95 raise ValueError(msg)
96 return super().__call__( # a subclass may add options of its own, as AsyncSoftFileLease does
97 **kwargs,
98 lock_file=lock_file,
99 timeout=timeout,
100 mode=mode,
101 thread_local=thread_local,
102 blocking=blocking,
103 is_singleton=is_singleton,
104 poll_interval=poll_interval,
105 lifetime=lifetime,
106 context_error_policy=context_error_policy,
107 close_error_policy=close_error_policy,
108 fallback_to_soft=fallback_to_soft,
109 preserve_lock_file=preserve_lock_file,
110 on_acquired=on_acquired,
111 loop=loop,
112 run_in_executor=run_in_executor,
113 executor=executor,
114 )
117class BaseAsyncFileLock(BaseFileLock, metaclass=AsyncFileLockMeta):
118 """
119 Base class for asynchronous file locks.
121 .. versionadded:: 3.15.0
123 """
125 _deadlock_holder_desc: str = "BaseAsyncFileLock instance in this task"
126 _constructor_lifetime_warning_stacklevel: int = 4
128 @staticmethod
129 def _deadlock_scope() -> Hashable | None:
130 # One event loop thread runs every task, so a thread-scoped registry cannot tell a same-task reacquire
131 # (a real deadlock: the polling task never reaches its own release) from another task queuing behind the
132 # holder (no deadlock: each poll yields, so the holder runs on and releases). Only the first may fail
133 # fast, so scope holders to the task.
134 return asyncio.current_task()
136 def __init__( # ruff:ignore[too-many-arguments] # public constructor: one parameter per documented lock option
137 self,
138 lock_file: str | os.PathLike[str],
139 timeout: float = -1,
140 mode: int = _UNSET_FILE_MODE,
141 thread_local: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument] # public API: positional bool kept for backwards compatibility
142 *,
143 blocking: bool = True,
144 is_singleton: bool = False,
145 poll_interval: float = 0.05,
146 lifetime: float | None = None,
147 context_error_policy: ContextErrorPolicy = "chain",
148 close_error_policy: CloseErrorPolicy = "default",
149 fallback_to_soft: bool = True,
150 preserve_lock_file: bool = False,
151 on_acquired: Callable[[int], None] | None = None,
152 loop: asyncio.AbstractEventLoop | None = None,
153 run_in_executor: bool = True,
154 executor: futures.Executor | None = None,
155 ) -> None:
156 """
157 Create a new lock object.
159 :param lock_file: path to the file
160 :param timeout: default timeout when acquiring the lock, in seconds. It will be used as fallback value in the
161 acquire method, if no timeout value (``None``) is given. If you want to disable the timeout, set it to a
162 negative value. A timeout of 0 means that there is exactly one attempt to acquire the file lock.
163 :param mode: file permissions for the lockfile. When not specified, the OS controls permissions via umask and
164 default ACLs, preserving POSIX default ACL inheritance in shared directories.
165 :param thread_local: Whether this object's internal context should be thread local or not. If this is set to
166 ``False`` then the lock will be reentrant across threads. When ``True`` (the default), **all fields of the
167 lock's internal context are per-thread**, including the configuration values ``poll_interval``, ``timeout``,
168 ``blocking``, ``mode``, and ``lifetime``. Setting one of these properties from one thread does not change
169 the value seen by another thread; threads that did not perform the write continue to see the value supplied
170 at construction time. If you need configuration values to be visible across threads, construct the lock
171 with ``thread_local=False``.
172 :param blocking: whether the lock should be blocking or not
173 :param is_singleton: If this is set to ``True`` then only one instance of this class will be created per lock
174 file. This is useful if you want to use the lock object for reentrant locking without needing to pass the
175 same object around.
176 :param poll_interval: default interval for polling the lock file, in seconds. It will be used as fallback value
177 in the acquire method, if no poll_interval value (``None``) is given.
178 :param lifetime: for :class:`AsyncSoftFileLock`, the age in seconds after which a waiting process may delete
179 the marker, even while its holder remains alive. This legacy expiry mode does not provide strict mutual
180 exclusion. ``None`` (the default) disables age-based expiry. Native OS locks (:class:`AsyncFileLock`)
181 cannot be revoked by file age and ignore a non-``None`` ``lifetime`` with a warning.
182 :param context_error_policy: how a context manager reconciles a failure in its body with a failure while
183 releasing on exit. ``"chain"`` (the default) keeps Python's behavior: the release error propagates with the
184 body error in its ``__context__``. ``"group"`` raises a :class:`BaseExceptionGroup` holding the body error
185 first and the release error second, so neither hides the other.
186 :param close_error_policy: for native locks (:class:`AsyncFileLock`), what to do with an ``os.close`` failure
187 after the OS unlock has already committed. ``"default"`` keeps each platform's historical behavior,
188 ``"raise"`` always propagates the ``OSError``, and ``"suppress"`` always ignores it.
189 :param fallback_to_soft: for :class:`AsyncFileLock`, whether to fall back to soft existence locking when
190 ``flock`` returns ``ENOSYS``. ``True`` (default) keeps the fallback; ``False`` propagates the error.
191 :param preserve_lock_file: for native locks (:class:`AsyncFileLock`), whether filelock promises not to unlink
192 the lock pathname on release. ``False`` (default) keeps each backend's cleanup; ``True`` keeps a stable file
193 identity (Windows skips its unlink, Unix refuses the ``ENOSYS`` soft fallback). :class:`AsyncSoftFileLock`
194 rejects ``True``.
195 :param on_acquired: for native locks (:class:`AsyncFileLock`), a callable invoked with the borrowed lock
196 descriptor once per physical acquisition, after the lock is held but before
197 :meth:`~BaseAsyncFileLock.acquire` returns. With ``run_in_executor=True`` (the default) it runs in the
198 backend executor. It must not close or unlock the descriptor; a raise rolls the acquisition back.
199 :class:`AsyncSoftFileLock` rejects it.
200 :param loop: The event loop to use. If not specified, the running event loop will be used.
201 :param run_in_executor: If this is set to ``True`` then the lock will be acquired in an executor.
202 :param executor: The executor to use. If not specified, the default executor will be used.
204 """
205 self._creator_pid = os.getpid()
206 self._is_thread_local = thread_local
207 self._is_singleton = is_singleton
208 self._context_error_policy = context_error_policy # already validated by the metaclass
209 self._close_error_policy = close_error_policy # already validated by the metaclass
210 self._fallback_to_soft = fallback_to_soft
211 self._preserve_lock_file = preserve_lock_file # already validated by the metaclass
212 self._on_acquired = on_acquired # already validated by the metaclass
213 self._transition_gate: Final[_AsyncTransitionGate] = _AsyncTransitionGate()
215 self._context: AsyncFileLockContext = (AsyncThreadLocalFileContext if thread_local else AsyncFileLockContext)(
216 lock_file=os.fspath(lock_file),
217 timeout=timeout,
218 mode=mode,
219 blocking=blocking,
220 poll_interval=poll_interval,
221 lifetime=lifetime,
222 loop=loop,
223 run_in_executor=run_in_executor,
224 executor=executor,
225 )
226 _register_fork_object(self)
228 @property
229 def run_in_executor(self) -> bool:
230 """Whether run in executor."""
231 return self._context.run_in_executor
233 @property
234 def executor(self) -> futures.Executor | None:
235 """The executor."""
236 return self._context.executor
238 @executor.setter
239 def executor(self, value: futures.Executor | None) -> None: # pragma: no cover
240 """
241 Change the executor.
243 :param futures.Executor | None value: the new executor or ``None``
245 """
246 self._context.executor = value
248 @property
249 def loop(self) -> asyncio.AbstractEventLoop | None:
250 """The event loop."""
251 return self._context.loop
253 async def acquire( # ty: ignore[invalid-method-override]
254 self,
255 timeout: float | None = None,
256 poll_interval: float | None = None,
257 *,
258 blocking: bool | None = None,
259 cancel_check: Callable[[], bool] | None = None,
260 ) -> AsyncAcquireReturnProxy:
261 """
262 Try to acquire the file lock.
264 :param timeout: maximum wait time for acquiring the lock, ``None`` means use the default
265 :attr:`~BaseFileLock.timeout` is and if ``timeout < 0``, there is no timeout and this method will block
266 until the lock could be acquired
267 :param poll_interval: interval of trying to acquire the lock file, ``None`` means use the default
268 :attr:`~BaseFileLock.poll_interval`
269 :param blocking: defaults to True. If False, function will return immediately if it cannot obtain a lock on the
270 first attempt. Otherwise, this method will block until the timeout expires or the lock is acquired.
271 :param cancel_check: a callable returning ``True`` when the acquisition should be canceled. Checked on each poll
272 iteration. When triggered, raises :class:`~Timeout` just like an expired timeout.
274 :returns: a context object that will unlock the file when the context is exited
276 :raises Timeout: if fails to acquire lock within the timeout period
278 .. code-block:: python
280 # You can use this method in the context manager (recommended)
281 with lock.acquire():
282 pass
284 # Or use an equivalent try-finally construct:
285 lock.acquire()
286 try:
287 pass
288 finally:
289 lock.release()
291 """
292 self._raise_if_inherited()
293 if timeout is None:
294 timeout = self._context.timeout
296 if blocking is None:
297 blocking = self._context.blocking
299 if poll_interval is None:
300 poll_interval = self._context.poll_interval
302 start_time = time.perf_counter()
303 try:
304 return await self._acquire_with_admission(
305 blocking=blocking,
306 cancel_check=cancel_check,
307 timeout=timeout,
308 poll_interval=poll_interval,
309 start_time=start_time,
310 )
311 except _AsyncTransitionUnavailableError:
312 raise Timeout(self.lock_file) from None
314 async def _acquire_with_admission(
315 self,
316 *,
317 blocking: bool,
318 cancel_check: Callable[[], bool] | None,
319 timeout: float,
320 poll_interval: float,
321 start_time: float,
322 ) -> AsyncAcquireReturnProxy:
323 async with self._transition_gate.hold_for_acquire(
324 blocking=blocking,
325 cancel_check=cancel_check,
326 deadline=None if timeout < 0 else start_time + timeout,
327 poll_interval=poll_interval,
328 ):
329 # A canceled provisional acquire must finish rollback before another caller can claim its descriptor.
330 canonical = _canonical(self.lock_file)
331 self._context.lock_counter += 1
332 self._raise_if_would_deadlock(canonical, timeout=timeout, blocking=blocking)
333 self._context.claim_root = canonical
334 try:
335 await self._async_poll_until_acquired(
336 blocking=blocking,
337 cancel_check=cancel_check,
338 timeout=timeout,
339 poll_interval=poll_interval,
340 start_time=start_time,
341 )
342 except BaseException:
343 self._reconcile_failed_acquire(canonical)
344 raise
345 finally:
346 self._context.claim_root = None
347 self._commit_acquire(canonical)
348 return AsyncAcquireReturnProxy(lock=self)
350 async def _async_poll_until_acquired(
351 self,
352 *,
353 blocking: bool,
354 cancel_check: Callable[[], bool] | None,
355 timeout: float,
356 poll_interval: float,
357 start_time: float,
358 ) -> None:
359 lock_id = id(self)
360 lock_filename = self.lock_file
361 while True:
362 self._raise_if_inherited()
363 if not self.is_locked:
364 self._try_break_expired_lock()
365 _LOGGER.debug("Attempting to acquire lock %s on %s", lock_id, lock_filename)
366 await self._run_acquire_attempt()
367 self._raise_if_inherited()
368 if self.is_locked:
369 _LOGGER.debug("Lock %s acquired on %s", lock_id, lock_filename)
370 return
371 if self._check_give_up(
372 blocking=blocking,
373 cancel_check=cancel_check,
374 timeout=timeout,
375 start_time=start_time,
376 ):
377 raise Timeout(lock_filename)
378 _LOGGER.debug("Lock %s not acquired on %s, waiting %s seconds ...", lock_id, lock_filename, poll_interval)
379 await asyncio.sleep(poll_interval)
381 async def _run_acquire_attempt(self) -> None:
382 acquire_future = self._start_internal_method(
383 self._acquire_with_fork_tracking_async
384 if iscoroutinefunction(self._acquire)
385 else self._acquire_with_fork_tracking
386 )
387 try:
388 await _wait_until_done(acquire_future)
389 except asyncio.CancelledError as cancellation:
390 acquire_error: BaseException | None = None
391 try:
392 await _drain_future(acquire_future)
393 except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below
394 acquire_error = error
396 rollback_error: BaseException | None = None
397 if self.is_locked:
398 try:
399 await _drain_future(self._start_tracked_release())
400 except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below
401 rollback_error = error
403 if acquire_error is not None:
404 if rollback_error is not None: # pragma: needs fcntl
405 self._raise_cancelled_errors(
406 "lock acquisition cancellation, backend attempt, and rollback failed",
407 cancellation,
408 acquire_error,
409 rollback_error,
410 )
411 self._raise_cancelled_errors(
412 "lock acquisition cancellation and backend attempt both failed", cancellation, acquire_error
413 )
414 if rollback_error is not None:
415 self._raise_cancelled_errors(
416 "lock acquisition cancellation and rollback both failed", cancellation, rollback_error
417 )
418 raise
419 try:
420 _future_result(acquire_future)
421 except asyncio.CancelledError as acquire_error:
422 await self._rollback_backend_cancelled_acquire(acquire_error)
424 async def _rollback_backend_cancelled_acquire(self, acquire_error: asyncio.CancelledError) -> NoReturn:
425 if self.is_locked:
426 try:
427 await _drain_future(self._start_tracked_release())
428 except BaseException as rollback_error: # ruff:ignore[blind-except] # both backend errors must surface
429 self._raise_acquire_rollback_errors(acquire_error, rollback_error)
430 raise acquire_error
432 def _raise_acquire_rollback_errors(self, acquire_error: BaseException, rollback_error: BaseException) -> NoReturn:
433 if self._context_error_policy == "group":
434 _raise_grouped_errors("lock acquisition backend and rollback both failed", acquire_error, rollback_error)
435 _raise_chained_errors(acquire_error, rollback_error)
437 def _raise_cancelled_errors(
438 self,
439 message: str,
440 cancellation: asyncio.CancelledError,
441 first_error: BaseException,
442 second_error: BaseException | None = None,
443 ) -> NoReturn:
444 if self._context_error_policy == "group":
445 marker = (
446 (_ASYNC_RELEASE_CANCELLATION_MARKER_ATTR, _ASYNC_RELEASE_CANCELLATION_MARKER)
447 if message == _ASYNC_RELEASE_CANCELLATION_ERRORS
448 else None
449 )
450 if second_error is None:
451 _raise_grouped_errors(message, cancellation, first_error, marker=marker)
452 _raise_grouped_errors(message, cancellation, first_error, second_error, marker=marker)
453 if (context := first_error.__context__) is not None and context is not cancellation:
454 if (cancellation_context := cancellation.__context__) is not None:
455 _append_exception_context(context, cancellation_context)
456 cancellation.__context__ = context
457 first_error.__context__ = cancellation
458 _raise_chained_errors(first_error, second_error)
460 async def release(self, force: bool = False) -> None: # ty: ignore[invalid-method-override] # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument] # public API: positional bool kept for backwards compatibility
461 """
462 Release the file lock. The lock is only completely released when the lock counter reaches 0. The lock file
463 itself may be deleted automatically, the behavior is platform-specific.
465 :param force: If true, the lock counter is ignored and the lock is released in every case.
467 """
468 if self._creator_pid != os.getpid() or not self.is_locked:
469 return
470 async with self._transition_gate.hold():
471 await self._release_serialized(force=force)
473 async def _release_serialized(self, *, force: bool) -> None:
474 if not self.is_locked:
475 return
476 if not force and self._context.lock_counter > 1:
477 self._context.lock_counter -= 1
478 return
480 lock_id, lock_filename = id(self), self.lock_file
481 _LOGGER.debug("Attempting to release lock %s on %s", lock_id, lock_filename)
482 release_future = self._start_tracked_release()
483 try:
484 await _wait_until_done(release_future)
485 except asyncio.CancelledError as cancellation:
486 try:
487 await _drain_future(release_future)
488 except BaseException as release_error: # ruff:ignore[blind-except] # cancellation and backend failure must both surface
489 self._commit_release_if_released()
490 self._raise_cancelled_errors(_ASYNC_RELEASE_CANCELLATION_ERRORS, cancellation, release_error)
491 self._commit_release()
492 raise
493 try:
494 _future_result(release_future)
495 except BaseException: # state follows the backend for control-flow exceptions too
496 self._commit_release_if_released()
497 raise
498 self._commit_release()
499 _LOGGER.debug("Lock %s released on %s", lock_id, lock_filename)
501 def _commit_release_if_released(self) -> None:
502 # Commit only when the backend actually unlocked (close or unlink failed after the OS unlock). If the lock is
503 # still held, keep the counter so a later release can retry.
504 if not self.is_locked:
505 self._commit_release()
507 def _start_internal_method(
508 self, method: Callable[[], None] | Callable[[], Coroutine[None, None, None]]
509 ) -> asyncio.Future[_BackendOutcome[None]]:
510 if iscoroutinefunction(method):
511 return asyncio.create_task(_capture_awaitable(cast("Callable[[], Coroutine[None, None, None]]", method)()))
512 loop = asyncio.get_running_loop()
513 sync_method = cast("Callable[[], None]", method)
514 if self.run_in_executor:
515 return loop.run_in_executor(self.executor, _capture_call, sync_method)
516 future: asyncio.Future[_BackendOutcome[None]] = loop.create_future()
517 future.set_result(_capture_call(sync_method))
518 return future
520 def _start_tracked_release(self) -> asyncio.Future[_BackendOutcome[None]]:
521 return self._start_internal_method(
522 self._release_with_fork_tracking_async
523 if iscoroutinefunction(self._release)
524 else self._release_with_fork_tracking
525 )
527 async def _acquire_with_fork_tracking_async(self) -> None:
528 with _fork_transition(self):
529 try:
530 await cast("Callable[[], Awaitable[None]]", self._acquire)()
531 except BaseException as acquisition_error:
532 await self._rollback_failed_acquire_async(acquisition_error)
533 raise
534 try:
535 self._register_context_descriptor()
536 except BaseException as registration_error: # cancellation must roll back the descriptor
537 await self._rollback_failed_registration_async(registration_error)
538 raise
539 if self.is_locked:
540 await self._invoke_on_acquired_async()
542 async def _rollback_failed_acquire_async(self, acquisition_error: BaseException) -> None:
543 if not self.is_locked:
544 return
545 registration_error: BaseException | None = None
546 tracking_error: BaseException | None = None
547 try:
548 self._register_context_descriptor()
549 except BaseException as error: # ruff:ignore[blind-except] # preserve registration and acquisition failures
550 registration_error = error
551 try:
552 # Rollback may fail too; retain the fd so a child can close it without another identity probe.
553 self._register_unverified_context_descriptor()
554 except BaseException as error: # ruff:ignore[blind-except] # pragma: no cover - allocation/control-flow during fallback
555 tracking_error = error
556 try:
557 await _drain_future(self._start_tracked_release())
558 except BaseException as rollback_error: # ruff:ignore[blind-except] # preserve rollback and acquisition failures
559 if registration_error is None and tracking_error is None:
560 self._raise_acquire_rollback_errors(acquisition_error, rollback_error)
561 _raise_cleanup_errors(
562 "lock acquisition cleanup failed",
563 acquisition_error,
564 registration_error,
565 tracking_error,
566 rollback_error,
567 )
568 if registration_error is not None:
569 _raise_cleanup_errors(
570 "lock acquisition cleanup failed", acquisition_error, registration_error, tracking_error
571 )
573 async def _rollback_failed_registration_async(self, registration_error: BaseException) -> None:
574 tracking_error: BaseException | None = None
575 try:
576 # Rollback may fail too; retain the fd so a child can close it without another identity probe.
577 self._register_unverified_context_descriptor()
578 except BaseException as error: # ruff:ignore[blind-except] # pragma: no cover - allocation/control-flow during fallback
579 tracking_error = error
580 try:
581 await _drain_future(self._start_tracked_release())
582 except BaseException as rollback_error: # ruff:ignore[blind-except] # preserve rollback and registration failures
583 _raise_cleanup_errors(
584 "descriptor registration cleanup failed", registration_error, tracking_error, rollback_error
585 )
586 if tracking_error is not None: # pragma: no cover - requires failed in-memory fallback
587 _raise_cleanup_errors("descriptor registration cleanup failed", registration_error, tracking_error)
589 async def _invoke_on_acquired_async(self) -> None:
590 if self._on_acquired is None or self._context.lock_counter != 1:
591 return
592 try:
593 self._on_acquired(cast("int", self._context.lock_file_fd))
594 except BaseException as callback_error: # caller control-flow errors must release the lock
595 callback_context = callback_error.__context__
596 try:
597 await _drain_future(self._start_tracked_release())
598 except BaseException as release_error: # ruff:ignore[blind-except] # both errors surface via the group below
599 _raise_body_and_release(callback_error, release_error)
600 callback_error.__context__ = callback_context
601 raise
603 async def _release_with_fork_tracking_async(self) -> None:
604 with _fork_transition(self):
605 try:
606 await cast("Callable[[], Awaitable[None]]", self._release)()
607 finally:
608 self._unregister_released_descriptor()
610 def __enter__(self) -> NoReturn:
611 """Sync context manager entry is not supported because lock acquisition is a coroutine."""
612 msg = "Use `async with`: acquire/release are coroutines and cannot be awaited in a sync context manager."
613 raise NotImplementedError(msg)
615 def __exit__(
616 self,
617 exc_type: type[BaseException] | None,
618 exc_value: BaseException | None,
619 traceback: TracebackType | None,
620 ) -> None:
621 """Sync context manager exit is not supported because lock release is a coroutine."""
622 msg = "Use `async with`: acquire/release are coroutines and cannot be awaited in a sync context manager."
623 raise NotImplementedError(msg)
625 async def __aenter__(self) -> Self:
626 """
627 Acquire the lock.
629 :returns: the lock object
631 """
632 await self.acquire()
633 return self
635 async def __aexit__(
636 self,
637 exc_type: type[BaseException] | None,
638 exc_value: BaseException | None,
639 traceback: TracebackType | None,
640 ) -> None:
641 """
642 Release the lock, reconciling a release failure with any body failure per :attr:`context_error_policy`.
644 :param exc_type: the exception type if raised
645 :param exc_value: the exception value if raised
646 :param traceback: the exception traceback if raised
648 """
649 await self._release_in_context(exc_value)
651 async def _release_in_context( # ty: ignore[invalid-method-override]
652 self, body_error: BaseException | None
653 ) -> None:
654 # The async counterpart of BaseFileLock._release_in_context: await release, then apply the same policy.
655 try:
656 await self.release()
657 except BaseException as release_error:
658 if body_error is None:
659 raise
660 if self._context_error_policy == "chain":
661 _append_exception_context(release_error, body_error)
662 raise
663 if (
664 errors := _grouped_errors(
665 release_error,
666 _ASYNC_RELEASE_CANCELLATION_ERRORS,
667 (_ASYNC_RELEASE_CANCELLATION_MARKER_ATTR, _ASYNC_RELEASE_CANCELLATION_MARKER),
668 )
669 ) is not None:
670 match errors:
671 # The marker is only ever set on a (CancelledError, backend_error) pair, so this always matches.
672 case (asyncio.CancelledError() as cancellation, backend_error): # pragma: no branch
673 _raise_grouped_errors(_ASYNC_CONTEXT_RELEASE_ERRORS, body_error, cancellation, backend_error)
674 _raise_body_and_release(body_error, release_error)
676 def __del__(self) -> None:
677 """Release on deletion; safe to call during GC even when no event loop is running."""
678 if vars(self).get("_creator_pid") != os.getpid():
679 return # pragma: forked child
680 with contextlib.suppress(Exception):
681 try:
682 loop = asyncio.get_running_loop()
683 except RuntimeError:
684 loop = self._context.loop if self._context.loop and not self._context.loop.is_closed() else None
685 if loop is None:
686 return
687 if not loop.is_running(): # pragma: no cover
688 loop.run_until_complete(self.release(force=True))
689 else:
690 loop.create_task(self.release(force=True))
693@dataclass
694class AsyncFileLockContext(FileLockContext):
695 """A dataclass which holds the context for a ``BaseAsyncFileLock`` object."""
697 #: Whether run in executor
698 run_in_executor: bool = True
700 #: The executor
701 executor: futures.Executor | None = None
703 #: The loop
704 loop: asyncio.AbstractEventLoop | None = None
707class AsyncThreadLocalFileContext(AsyncFileLockContext, local):
708 """A thread local version of the ``FileLockContext`` class."""
711class AsyncAcquireReturnProxy:
712 """A context-aware object that will release the lock file when exiting."""
714 def __init__(self, lock: BaseAsyncFileLock) -> None: # ruff:ignore[undocumented-public-init] # trivial release-on-exit proxy
715 self.lock = lock
717 async def __aenter__(self) -> BaseAsyncFileLock: # ruff:ignore[undocumented-magic-method] # returns the wrapped lock
718 return self.lock
720 async def __aexit__( # ruff:ignore[undocumented-magic-method] # releases the wrapped lock on exit
721 self,
722 exc_type: type[BaseException] | None,
723 exc_value: BaseException | None,
724 traceback: TracebackType | None,
725 ) -> None:
726 await self.lock._release_in_context(exc_value) # ruff:ignore[private-member-access] # releases the wrapped lock's context
729class AsyncSoftFileLock(SoftFileLock, BaseAsyncFileLock):
730 """Simply watches the existence of the lock file."""
732 _lifetime_replacements: tuple[str, str] | None = ("AsyncStrictSoftFileLock", "AsyncSoftFileLease")
735class AsyncStrictSoftFileLock(StrictSoftFileLock, BaseAsyncFileLock):
736 """Run strict owner-claim locking without blocking the event loop."""
739class AsyncSoftFileLease(SoftFileLease, BaseAsyncFileLock):
740 """Existence lock whose claim expires, so a peer may take it while the previous holder still runs."""
743class AsyncUnixFileLock(UnixFileLock, BaseAsyncFileLock):
744 """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems."""
747class AsyncWindowsFileLock(WindowsFileLock, BaseAsyncFileLock):
748 """Uses the :func:`msvcrt.locking` to hard lock the lock file on windows systems."""
751__all__ = [
752 "AsyncAcquireReturnProxy",
753 "AsyncSoftFileLease",
754 "AsyncSoftFileLock",
755 "AsyncStrictSoftFileLock",
756 "AsyncUnixFileLock",
757 "AsyncWindowsFileLock",
758 "BaseAsyncFileLock",
759]