Spaces:
Paused
Paused
| """An asyncio-based implementation of the file lock.""" | |
| from __future__ import annotations | |
| import asyncio | |
| import contextlib | |
| import logging | |
| import os | |
| import time | |
| from dataclasses import dataclass | |
| from inspect import iscoroutinefunction | |
| from threading import local | |
| from typing import TYPE_CHECKING, Final, NoReturn, TypeVar, cast | |
| from ._api import ( | |
| _UNSET_FILE_MODE, | |
| BaseFileLock, | |
| CloseErrorPolicy, | |
| ContextErrorPolicy, | |
| FileLockContext, | |
| FileLockMeta, | |
| _append_exception_context, | |
| _canonical, | |
| _ExtraValue, | |
| _fork_transition, | |
| _grouped_errors, | |
| _raise_body_and_release, | |
| _raise_chained_errors, | |
| _raise_cleanup_errors, | |
| _raise_grouped_errors, | |
| _register_fork_object, | |
| ) | |
| from ._async import ( | |
| _AsyncTransitionGate, | |
| _AsyncTransitionUnavailableError, | |
| _BackendOutcome, | |
| _capture_awaitable, | |
| _capture_call, | |
| _drain_future, | |
| _future_result, | |
| _wait_until_done, | |
| ) | |
| from ._error import Timeout | |
| from ._lease import SoftFileLease | |
| from ._soft import SoftFileLock | |
| from ._strict import StrictSoftFileLock | |
| from ._unix import UnixFileLock | |
| from ._windows import WindowsFileLock | |
| if TYPE_CHECKING: | |
| import sys | |
| from collections.abc import Awaitable, Callable, Coroutine, Hashable | |
| from concurrent import futures | |
| from types import TracebackType | |
| if sys.version_info >= (3, 11): # pragma: no cover (py311+) | |
| from typing import Self | |
| else: # pragma: no cover (<py311) | |
| from typing_extensions import Self | |
| _LOGGER: Final[logging.Logger] = logging.getLogger("filelock") | |
| _ASYNC_RELEASE_CANCELLATION_ERRORS: Final[str] = "lock release cancellation and backend release both failed" | |
| _ASYNC_CONTEXT_RELEASE_ERRORS: Final[str] = "context body, release cancellation, and backend release failed" | |
| _ASYNC_RELEASE_CANCELLATION_MARKER_ATTR: Final[str] = "_filelock_async_release_cancellation" | |
| _ASYNC_RELEASE_CANCELLATION_MARKER: Final[list[None]] = [] | |
| _AT = TypeVar("_AT", bound="BaseAsyncFileLock") | |
| class AsyncFileLockMeta(FileLockMeta): | |
| def __call__( # ruff:ignore[too-many-arguments] # forwards the public constructor's documented parameters | |
| cls: type[_AT], # ruff:ignore[invalid-first-argument-name-for-method] # metaclass __call__ receives the class being constructed | |
| lock_file: str | os.PathLike[str], | |
| timeout: float = -1, | |
| mode: int = _UNSET_FILE_MODE, | |
| thread_local: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument] # public API: positional bool kept for backwards compatibility | |
| *, | |
| blocking: bool = True, | |
| is_singleton: bool = False, | |
| poll_interval: float = 0.05, | |
| lifetime: float | None = None, | |
| context_error_policy: ContextErrorPolicy = "chain", | |
| close_error_policy: CloseErrorPolicy = "default", | |
| fallback_to_soft: bool = True, | |
| preserve_lock_file: bool = False, | |
| on_acquired: Callable[[int], None] | None = None, | |
| loop: asyncio.AbstractEventLoop | None = None, | |
| run_in_executor: bool = True, | |
| executor: futures.Executor | None = None, | |
| **kwargs: _ExtraValue, | |
| ) -> _AT: | |
| if thread_local and run_in_executor: | |
| msg = "run_in_executor is not supported when thread_local is True" | |
| raise ValueError(msg) | |
| return super().__call__( # a subclass may add options of its own, as AsyncSoftFileLease does | |
| **kwargs, | |
| lock_file=lock_file, | |
| timeout=timeout, | |
| mode=mode, | |
| thread_local=thread_local, | |
| blocking=blocking, | |
| is_singleton=is_singleton, | |
| poll_interval=poll_interval, | |
| lifetime=lifetime, | |
| context_error_policy=context_error_policy, | |
| close_error_policy=close_error_policy, | |
| fallback_to_soft=fallback_to_soft, | |
| preserve_lock_file=preserve_lock_file, | |
| on_acquired=on_acquired, | |
| loop=loop, | |
| run_in_executor=run_in_executor, | |
| executor=executor, | |
| ) | |
| class BaseAsyncFileLock(BaseFileLock, metaclass=AsyncFileLockMeta): | |
| """ | |
| Base class for asynchronous file locks. | |
| .. versionadded:: 3.15.0 | |
| """ | |
| _deadlock_holder_desc: str = "BaseAsyncFileLock instance in this task" | |
| _constructor_lifetime_warning_stacklevel: int = 4 | |
| def _deadlock_scope() -> Hashable | None: | |
| # One event loop thread runs every task, so a thread-scoped registry cannot tell a same-task reacquire | |
| # (a real deadlock: the polling task never reaches its own release) from another task queuing behind the | |
| # holder (no deadlock: each poll yields, so the holder runs on and releases). Only the first may fail | |
| # fast, so scope holders to the task. | |
| return asyncio.current_task() | |
| def __init__( # ruff:ignore[too-many-arguments] # public constructor: one parameter per documented lock option | |
| self, | |
| lock_file: str | os.PathLike[str], | |
| timeout: float = -1, | |
| mode: int = _UNSET_FILE_MODE, | |
| thread_local: bool = False, # ruff:ignore[boolean-type-hint-positional-argument, boolean-default-value-positional-argument] # public API: positional bool kept for backwards compatibility | |
| *, | |
| blocking: bool = True, | |
| is_singleton: bool = False, | |
| poll_interval: float = 0.05, | |
| lifetime: float | None = None, | |
| context_error_policy: ContextErrorPolicy = "chain", | |
| close_error_policy: CloseErrorPolicy = "default", | |
| fallback_to_soft: bool = True, | |
| preserve_lock_file: bool = False, | |
| on_acquired: Callable[[int], None] | None = None, | |
| loop: asyncio.AbstractEventLoop | None = None, | |
| run_in_executor: bool = True, | |
| executor: futures.Executor | None = None, | |
| ) -> None: | |
| """ | |
| Create a new lock object. | |
| :param lock_file: path to the file | |
| :param timeout: default timeout when acquiring the lock, in seconds. It will be used as fallback value in the | |
| acquire method, if no timeout value (``None``) is given. If you want to disable the timeout, set it to a | |
| negative value. A timeout of 0 means that there is exactly one attempt to acquire the file lock. | |
| :param mode: file permissions for the lockfile. When not specified, the OS controls permissions via umask and | |
| default ACLs, preserving POSIX default ACL inheritance in shared directories. | |
| :param thread_local: Whether this object's internal context should be thread local or not. If this is set to | |
| ``False`` then the lock will be reentrant across threads. When ``True`` (the default), **all fields of the | |
| lock's internal context are per-thread**, including the configuration values ``poll_interval``, ``timeout``, | |
| ``blocking``, ``mode``, and ``lifetime``. Setting one of these properties from one thread does not change | |
| the value seen by another thread; threads that did not perform the write continue to see the value supplied | |
| at construction time. If you need configuration values to be visible across threads, construct the lock | |
| with ``thread_local=False``. | |
| :param blocking: whether the lock should be blocking or not | |
| :param is_singleton: If this is set to ``True`` then only one instance of this class will be created per lock | |
| file. This is useful if you want to use the lock object for reentrant locking without needing to pass the | |
| same object around. | |
| :param poll_interval: default interval for polling the lock file, in seconds. It will be used as fallback value | |
| in the acquire method, if no poll_interval value (``None``) is given. | |
| :param lifetime: for :class:`AsyncSoftFileLock`, the age in seconds after which a waiting process may delete | |
| the marker, even while its holder remains alive. This legacy expiry mode does not provide strict mutual | |
| exclusion. ``None`` (the default) disables age-based expiry. Native OS locks (:class:`AsyncFileLock`) | |
| cannot be revoked by file age and ignore a non-``None`` ``lifetime`` with a warning. | |
| :param context_error_policy: how a context manager reconciles a failure in its body with a failure while | |
| releasing on exit. ``"chain"`` (the default) keeps Python's behavior: the release error propagates with the | |
| body error in its ``__context__``. ``"group"`` raises a :class:`BaseExceptionGroup` holding the body error | |
| first and the release error second, so neither hides the other. | |
| :param close_error_policy: for native locks (:class:`AsyncFileLock`), what to do with an ``os.close`` failure | |
| after the OS unlock has already committed. ``"default"`` keeps each platform's historical behavior, | |
| ``"raise"`` always propagates the ``OSError``, and ``"suppress"`` always ignores it. | |
| :param fallback_to_soft: for :class:`AsyncFileLock`, whether to fall back to soft existence locking when | |
| ``flock`` returns ``ENOSYS``. ``True`` (default) keeps the fallback; ``False`` propagates the error. | |
| :param preserve_lock_file: for native locks (:class:`AsyncFileLock`), whether filelock promises not to unlink | |
| the lock pathname on release. ``False`` (default) keeps each backend's cleanup; ``True`` keeps a stable file | |
| identity (Windows skips its unlink, Unix refuses the ``ENOSYS`` soft fallback). :class:`AsyncSoftFileLock` | |
| rejects ``True``. | |
| :param on_acquired: for native locks (:class:`AsyncFileLock`), a callable invoked with the borrowed lock | |
| descriptor once per physical acquisition, after the lock is held but before | |
| :meth:`~BaseAsyncFileLock.acquire` returns. With ``run_in_executor=True`` (the default) it runs in the | |
| backend executor. It must not close or unlock the descriptor; a raise rolls the acquisition back. | |
| :class:`AsyncSoftFileLock` rejects it. | |
| :param loop: The event loop to use. If not specified, the running event loop will be used. | |
| :param run_in_executor: If this is set to ``True`` then the lock will be acquired in an executor. | |
| :param executor: The executor to use. If not specified, the default executor will be used. | |
| """ | |
| self._creator_pid = os.getpid() | |
| self._is_thread_local = thread_local | |
| self._is_singleton = is_singleton | |
| self._context_error_policy = context_error_policy # already validated by the metaclass | |
| self._close_error_policy = close_error_policy # already validated by the metaclass | |
| self._fallback_to_soft = fallback_to_soft | |
| self._preserve_lock_file = preserve_lock_file # already validated by the metaclass | |
| self._on_acquired = on_acquired # already validated by the metaclass | |
| self._transition_gate: Final[_AsyncTransitionGate] = _AsyncTransitionGate() | |
| self._context: AsyncFileLockContext = (AsyncThreadLocalFileContext if thread_local else AsyncFileLockContext)( | |
| lock_file=os.fspath(lock_file), | |
| timeout=timeout, | |
| mode=mode, | |
| blocking=blocking, | |
| poll_interval=poll_interval, | |
| lifetime=lifetime, | |
| loop=loop, | |
| run_in_executor=run_in_executor, | |
| executor=executor, | |
| ) | |
| _register_fork_object(self) | |
| def run_in_executor(self) -> bool: | |
| """Whether run in executor.""" | |
| return self._context.run_in_executor | |
| def executor(self) -> futures.Executor | None: | |
| """The executor.""" | |
| return self._context.executor | |
| def executor(self, value: futures.Executor | None) -> None: # pragma: no cover | |
| """ | |
| Change the executor. | |
| :param futures.Executor | None value: the new executor or ``None`` | |
| """ | |
| self._context.executor = value | |
| def loop(self) -> asyncio.AbstractEventLoop | None: | |
| """The event loop.""" | |
| return self._context.loop | |
| async def acquire( # ty: ignore[invalid-method-override] | |
| self, | |
| timeout: float | None = None, | |
| poll_interval: float | None = None, | |
| *, | |
| blocking: bool | None = None, | |
| cancel_check: Callable[[], bool] | None = None, | |
| ) -> AsyncAcquireReturnProxy: | |
| """ | |
| Try to acquire the file lock. | |
| :param timeout: maximum wait time for acquiring the lock, ``None`` means use the default | |
| :attr:`~BaseFileLock.timeout` is and if ``timeout < 0``, there is no timeout and this method will block | |
| until the lock could be acquired | |
| :param poll_interval: interval of trying to acquire the lock file, ``None`` means use the default | |
| :attr:`~BaseFileLock.poll_interval` | |
| :param blocking: defaults to True. If False, function will return immediately if it cannot obtain a lock on the | |
| first attempt. Otherwise, this method will block until the timeout expires or the lock is acquired. | |
| :param cancel_check: a callable returning ``True`` when the acquisition should be canceled. Checked on each poll | |
| iteration. When triggered, raises :class:`~Timeout` just like an expired timeout. | |
| :returns: a context object that will unlock the file when the context is exited | |
| :raises Timeout: if fails to acquire lock within the timeout period | |
| .. code-block:: python | |
| # You can use this method in the context manager (recommended) | |
| with lock.acquire(): | |
| pass | |
| # Or use an equivalent try-finally construct: | |
| lock.acquire() | |
| try: | |
| pass | |
| finally: | |
| lock.release() | |
| """ | |
| self._raise_if_inherited() | |
| if timeout is None: | |
| timeout = self._context.timeout | |
| if blocking is None: | |
| blocking = self._context.blocking | |
| if poll_interval is None: | |
| poll_interval = self._context.poll_interval | |
| start_time = time.perf_counter() | |
| try: | |
| return await self._acquire_with_admission( | |
| blocking=blocking, | |
| cancel_check=cancel_check, | |
| timeout=timeout, | |
| poll_interval=poll_interval, | |
| start_time=start_time, | |
| ) | |
| except _AsyncTransitionUnavailableError: | |
| raise Timeout(self.lock_file) from None | |
| async def _acquire_with_admission( | |
| self, | |
| *, | |
| blocking: bool, | |
| cancel_check: Callable[[], bool] | None, | |
| timeout: float, | |
| poll_interval: float, | |
| start_time: float, | |
| ) -> AsyncAcquireReturnProxy: | |
| async with self._transition_gate.hold_for_acquire( | |
| blocking=blocking, | |
| cancel_check=cancel_check, | |
| deadline=None if timeout < 0 else start_time + timeout, | |
| poll_interval=poll_interval, | |
| ): | |
| # A canceled provisional acquire must finish rollback before another caller can claim its descriptor. | |
| canonical = _canonical(self.lock_file) | |
| self._context.lock_counter += 1 | |
| self._raise_if_would_deadlock(canonical, timeout=timeout, blocking=blocking) | |
| self._context.claim_root = canonical | |
| try: | |
| await self._async_poll_until_acquired( | |
| blocking=blocking, | |
| cancel_check=cancel_check, | |
| timeout=timeout, | |
| poll_interval=poll_interval, | |
| start_time=start_time, | |
| ) | |
| except BaseException: | |
| self._reconcile_failed_acquire(canonical) | |
| raise | |
| finally: | |
| self._context.claim_root = None | |
| self._commit_acquire(canonical) | |
| return AsyncAcquireReturnProxy(lock=self) | |
| async def _async_poll_until_acquired( | |
| self, | |
| *, | |
| blocking: bool, | |
| cancel_check: Callable[[], bool] | None, | |
| timeout: float, | |
| poll_interval: float, | |
| start_time: float, | |
| ) -> None: | |
| lock_id = id(self) | |
| lock_filename = self.lock_file | |
| while True: | |
| self._raise_if_inherited() | |
| if not self.is_locked: | |
| self._try_break_expired_lock() | |
| _LOGGER.debug("Attempting to acquire lock %s on %s", lock_id, lock_filename) | |
| await self._run_acquire_attempt() | |
| self._raise_if_inherited() | |
| if self.is_locked: | |
| _LOGGER.debug("Lock %s acquired on %s", lock_id, lock_filename) | |
| return | |
| if self._check_give_up( | |
| blocking=blocking, | |
| cancel_check=cancel_check, | |
| timeout=timeout, | |
| start_time=start_time, | |
| ): | |
| raise Timeout(lock_filename) | |
| _LOGGER.debug("Lock %s not acquired on %s, waiting %s seconds ...", lock_id, lock_filename, poll_interval) | |
| await asyncio.sleep(poll_interval) | |
| async def _run_acquire_attempt(self) -> None: | |
| acquire_future = self._start_internal_method( | |
| self._acquire_with_fork_tracking_async | |
| if iscoroutinefunction(self._acquire) | |
| else self._acquire_with_fork_tracking | |
| ) | |
| try: | |
| await _wait_until_done(acquire_future) | |
| except asyncio.CancelledError as cancellation: | |
| acquire_error: BaseException | None = None | |
| try: | |
| await _drain_future(acquire_future) | |
| except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below | |
| acquire_error = error | |
| rollback_error: BaseException | None = None | |
| if self.is_locked: | |
| try: | |
| await _drain_future(self._start_tracked_release()) | |
| except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below | |
| rollback_error = error | |
| if acquire_error is not None: | |
| if rollback_error is not None: # pragma: needs fcntl | |
| self._raise_cancelled_errors( | |
| "lock acquisition cancellation, backend attempt, and rollback failed", | |
| cancellation, | |
| acquire_error, | |
| rollback_error, | |
| ) | |
| self._raise_cancelled_errors( | |
| "lock acquisition cancellation and backend attempt both failed", cancellation, acquire_error | |
| ) | |
| if rollback_error is not None: | |
| self._raise_cancelled_errors( | |
| "lock acquisition cancellation and rollback both failed", cancellation, rollback_error | |
| ) | |
| raise | |
| try: | |
| _future_result(acquire_future) | |
| except asyncio.CancelledError as acquire_error: | |
| await self._rollback_backend_cancelled_acquire(acquire_error) | |
| async def _rollback_backend_cancelled_acquire(self, acquire_error: asyncio.CancelledError) -> NoReturn: | |
| if self.is_locked: | |
| try: | |
| await _drain_future(self._start_tracked_release()) | |
| except BaseException as rollback_error: # ruff:ignore[blind-except] # both backend errors must surface | |
| self._raise_acquire_rollback_errors(acquire_error, rollback_error) | |
| raise acquire_error | |
| def _raise_acquire_rollback_errors(self, acquire_error: BaseException, rollback_error: BaseException) -> NoReturn: | |
| if self._context_error_policy == "group": | |
| _raise_grouped_errors("lock acquisition backend and rollback both failed", acquire_error, rollback_error) | |
| _raise_chained_errors(acquire_error, rollback_error) | |
| def _raise_cancelled_errors( | |
| self, | |
| message: str, | |
| cancellation: asyncio.CancelledError, | |
| first_error: BaseException, | |
| second_error: BaseException | None = None, | |
| ) -> NoReturn: | |
| if self._context_error_policy == "group": | |
| marker = ( | |
| (_ASYNC_RELEASE_CANCELLATION_MARKER_ATTR, _ASYNC_RELEASE_CANCELLATION_MARKER) | |
| if message == _ASYNC_RELEASE_CANCELLATION_ERRORS | |
| else None | |
| ) | |
| if second_error is None: | |
| _raise_grouped_errors(message, cancellation, first_error, marker=marker) | |
| _raise_grouped_errors(message, cancellation, first_error, second_error, marker=marker) | |
| if (context := first_error.__context__) is not None and context is not cancellation: | |
| if (cancellation_context := cancellation.__context__) is not None: | |
| _append_exception_context(context, cancellation_context) | |
| cancellation.__context__ = context | |
| first_error.__context__ = cancellation | |
| _raise_chained_errors(first_error, second_error) | |
| 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 | |
| """ | |
| Release the file lock. The lock is only completely released when the lock counter reaches 0. The lock file | |
| itself may be deleted automatically, the behavior is platform-specific. | |
| :param force: If true, the lock counter is ignored and the lock is released in every case. | |
| """ | |
| if self._creator_pid != os.getpid() or not self.is_locked: | |
| return | |
| async with self._transition_gate.hold(): | |
| await self._release_serialized(force=force) | |
| async def _release_serialized(self, *, force: bool) -> None: | |
| if not self.is_locked: | |
| return | |
| if not force and self._context.lock_counter > 1: | |
| self._context.lock_counter -= 1 | |
| return | |
| lock_id, lock_filename = id(self), self.lock_file | |
| _LOGGER.debug("Attempting to release lock %s on %s", lock_id, lock_filename) | |
| release_future = self._start_tracked_release() | |
| try: | |
| await _wait_until_done(release_future) | |
| except asyncio.CancelledError as cancellation: | |
| try: | |
| await _drain_future(release_future) | |
| except BaseException as release_error: # ruff:ignore[blind-except] # cancellation and backend failure must both surface | |
| self._commit_release_if_released() | |
| self._raise_cancelled_errors(_ASYNC_RELEASE_CANCELLATION_ERRORS, cancellation, release_error) | |
| self._commit_release() | |
| raise | |
| try: | |
| _future_result(release_future) | |
| except BaseException: # state follows the backend for control-flow exceptions too | |
| self._commit_release_if_released() | |
| raise | |
| self._commit_release() | |
| _LOGGER.debug("Lock %s released on %s", lock_id, lock_filename) | |
| def _commit_release_if_released(self) -> None: | |
| # Commit only when the backend actually unlocked (close or unlink failed after the OS unlock). If the lock is | |
| # still held, keep the counter so a later release can retry. | |
| if not self.is_locked: | |
| self._commit_release() | |
| def _start_internal_method( | |
| self, method: Callable[[], None] | Callable[[], Coroutine[None, None, None]] | |
| ) -> asyncio.Future[_BackendOutcome[None]]: | |
| if iscoroutinefunction(method): | |
| return asyncio.create_task(_capture_awaitable(cast("Callable[[], Coroutine[None, None, None]]", method)())) | |
| loop = asyncio.get_running_loop() | |
| sync_method = cast("Callable[[], None]", method) | |
| if self.run_in_executor: | |
| return loop.run_in_executor(self.executor, _capture_call, sync_method) | |
| future: asyncio.Future[_BackendOutcome[None]] = loop.create_future() | |
| future.set_result(_capture_call(sync_method)) | |
| return future | |
| def _start_tracked_release(self) -> asyncio.Future[_BackendOutcome[None]]: | |
| return self._start_internal_method( | |
| self._release_with_fork_tracking_async | |
| if iscoroutinefunction(self._release) | |
| else self._release_with_fork_tracking | |
| ) | |
| async def _acquire_with_fork_tracking_async(self) -> None: | |
| with _fork_transition(self): | |
| try: | |
| await cast("Callable[[], Awaitable[None]]", self._acquire)() | |
| except BaseException as acquisition_error: | |
| await self._rollback_failed_acquire_async(acquisition_error) | |
| raise | |
| try: | |
| self._register_context_descriptor() | |
| except BaseException as registration_error: # cancellation must roll back the descriptor | |
| await self._rollback_failed_registration_async(registration_error) | |
| raise | |
| if self.is_locked: | |
| await self._invoke_on_acquired_async() | |
| async def _rollback_failed_acquire_async(self, acquisition_error: BaseException) -> None: | |
| if not self.is_locked: | |
| return | |
| registration_error: BaseException | None = None | |
| tracking_error: BaseException | None = None | |
| try: | |
| self._register_context_descriptor() | |
| except BaseException as error: # ruff:ignore[blind-except] # preserve registration and acquisition failures | |
| registration_error = error | |
| try: | |
| # Rollback may fail too; retain the fd so a child can close it without another identity probe. | |
| self._register_unverified_context_descriptor() | |
| except BaseException as error: # ruff:ignore[blind-except] # pragma: no cover - allocation/control-flow during fallback | |
| tracking_error = error | |
| try: | |
| await _drain_future(self._start_tracked_release()) | |
| except BaseException as rollback_error: # ruff:ignore[blind-except] # preserve rollback and acquisition failures | |
| if registration_error is None and tracking_error is None: | |
| self._raise_acquire_rollback_errors(acquisition_error, rollback_error) | |
| _raise_cleanup_errors( | |
| "lock acquisition cleanup failed", | |
| acquisition_error, | |
| registration_error, | |
| tracking_error, | |
| rollback_error, | |
| ) | |
| if registration_error is not None: | |
| _raise_cleanup_errors( | |
| "lock acquisition cleanup failed", acquisition_error, registration_error, tracking_error | |
| ) | |
| async def _rollback_failed_registration_async(self, registration_error: BaseException) -> None: | |
| tracking_error: BaseException | None = None | |
| try: | |
| # Rollback may fail too; retain the fd so a child can close it without another identity probe. | |
| self._register_unverified_context_descriptor() | |
| except BaseException as error: # ruff:ignore[blind-except] # pragma: no cover - allocation/control-flow during fallback | |
| tracking_error = error | |
| try: | |
| await _drain_future(self._start_tracked_release()) | |
| except BaseException as rollback_error: # ruff:ignore[blind-except] # preserve rollback and registration failures | |
| _raise_cleanup_errors( | |
| "descriptor registration cleanup failed", registration_error, tracking_error, rollback_error | |
| ) | |
| if tracking_error is not None: # pragma: no cover - requires failed in-memory fallback | |
| _raise_cleanup_errors("descriptor registration cleanup failed", registration_error, tracking_error) | |
| async def _invoke_on_acquired_async(self) -> None: | |
| if self._on_acquired is None or self._context.lock_counter != 1: | |
| return | |
| try: | |
| self._on_acquired(cast("int", self._context.lock_file_fd)) | |
| except BaseException as callback_error: # caller control-flow errors must release the lock | |
| callback_context = callback_error.__context__ | |
| try: | |
| await _drain_future(self._start_tracked_release()) | |
| except BaseException as release_error: # ruff:ignore[blind-except] # both errors surface via the group below | |
| _raise_body_and_release(callback_error, release_error) | |
| callback_error.__context__ = callback_context | |
| raise | |
| async def _release_with_fork_tracking_async(self) -> None: | |
| with _fork_transition(self): | |
| try: | |
| await cast("Callable[[], Awaitable[None]]", self._release)() | |
| finally: | |
| self._unregister_released_descriptor() | |
| def __enter__(self) -> NoReturn: | |
| """Sync context manager entry is not supported because lock acquisition is a coroutine.""" | |
| msg = "Use `async with`: acquire/release are coroutines and cannot be awaited in a sync context manager." | |
| raise NotImplementedError(msg) | |
| def __exit__( | |
| self, | |
| exc_type: type[BaseException] | None, | |
| exc_value: BaseException | None, | |
| traceback: TracebackType | None, | |
| ) -> None: | |
| """Sync context manager exit is not supported because lock release is a coroutine.""" | |
| msg = "Use `async with`: acquire/release are coroutines and cannot be awaited in a sync context manager." | |
| raise NotImplementedError(msg) | |
| async def __aenter__(self) -> Self: | |
| """ | |
| Acquire the lock. | |
| :returns: the lock object | |
| """ | |
| await self.acquire() | |
| return self | |
| async def __aexit__( | |
| self, | |
| exc_type: type[BaseException] | None, | |
| exc_value: BaseException | None, | |
| traceback: TracebackType | None, | |
| ) -> None: | |
| """ | |
| Release the lock, reconciling a release failure with any body failure per :attr:`context_error_policy`. | |
| :param exc_type: the exception type if raised | |
| :param exc_value: the exception value if raised | |
| :param traceback: the exception traceback if raised | |
| """ | |
| await self._release_in_context(exc_value) | |
| async def _release_in_context( # ty: ignore[invalid-method-override] | |
| self, body_error: BaseException | None | |
| ) -> None: | |
| # The async counterpart of BaseFileLock._release_in_context: await release, then apply the same policy. | |
| try: | |
| await self.release() | |
| except BaseException as release_error: | |
| if body_error is None: | |
| raise | |
| if self._context_error_policy == "chain": | |
| _append_exception_context(release_error, body_error) | |
| raise | |
| if ( | |
| errors := _grouped_errors( | |
| release_error, | |
| _ASYNC_RELEASE_CANCELLATION_ERRORS, | |
| (_ASYNC_RELEASE_CANCELLATION_MARKER_ATTR, _ASYNC_RELEASE_CANCELLATION_MARKER), | |
| ) | |
| ) is not None: | |
| match errors: | |
| # The marker is only ever set on a (CancelledError, backend_error) pair, so this always matches. | |
| case (asyncio.CancelledError() as cancellation, backend_error): # pragma: no branch | |
| _raise_grouped_errors(_ASYNC_CONTEXT_RELEASE_ERRORS, body_error, cancellation, backend_error) | |
| _raise_body_and_release(body_error, release_error) | |
| def __del__(self) -> None: | |
| """Release on deletion; safe to call during GC even when no event loop is running.""" | |
| if vars(self).get("_creator_pid") != os.getpid(): | |
| return # pragma: forked child | |
| with contextlib.suppress(Exception): | |
| try: | |
| loop = asyncio.get_running_loop() | |
| except RuntimeError: | |
| loop = self._context.loop if self._context.loop and not self._context.loop.is_closed() else None | |
| if loop is None: | |
| return | |
| if not loop.is_running(): # pragma: no cover | |
| loop.run_until_complete(self.release(force=True)) | |
| else: | |
| loop.create_task(self.release(force=True)) | |
| class AsyncFileLockContext(FileLockContext): | |
| """A dataclass which holds the context for a ``BaseAsyncFileLock`` object.""" | |
| #: Whether run in executor | |
| run_in_executor: bool = True | |
| #: The executor | |
| executor: futures.Executor | None = None | |
| #: The loop | |
| loop: asyncio.AbstractEventLoop | None = None | |
| class AsyncThreadLocalFileContext(AsyncFileLockContext, local): | |
| """A thread local version of the ``FileLockContext`` class.""" | |
| class AsyncAcquireReturnProxy: | |
| """A context-aware object that will release the lock file when exiting.""" | |
| def __init__(self, lock: BaseAsyncFileLock) -> None: # ruff:ignore[undocumented-public-init] # trivial release-on-exit proxy | |
| self.lock = lock | |
| async def __aenter__(self) -> BaseAsyncFileLock: # ruff:ignore[undocumented-magic-method] # returns the wrapped lock | |
| return self.lock | |
| async def __aexit__( # ruff:ignore[undocumented-magic-method] # releases the wrapped lock on exit | |
| self, | |
| exc_type: type[BaseException] | None, | |
| exc_value: BaseException | None, | |
| traceback: TracebackType | None, | |
| ) -> None: | |
| await self.lock._release_in_context(exc_value) # ruff:ignore[private-member-access] # releases the wrapped lock's context | |
| class AsyncSoftFileLock(SoftFileLock, BaseAsyncFileLock): | |
| """Simply watches the existence of the lock file.""" | |
| _lifetime_replacements: tuple[str, str] | None = ("AsyncStrictSoftFileLock", "AsyncSoftFileLease") | |
| class AsyncStrictSoftFileLock(StrictSoftFileLock, BaseAsyncFileLock): | |
| """Run strict owner-claim locking without blocking the event loop.""" | |
| class AsyncSoftFileLease(SoftFileLease, BaseAsyncFileLock): | |
| """Existence lock whose claim expires, so a peer may take it while the previous holder still runs.""" | |
| class AsyncUnixFileLock(UnixFileLock, BaseAsyncFileLock): | |
| """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems.""" | |
| class AsyncWindowsFileLock(WindowsFileLock, BaseAsyncFileLock): | |
| """Uses the :func:`msvcrt.locking` to hard lock the lock file on windows systems.""" | |
| __all__ = [ | |
| "AsyncAcquireReturnProxy", | |
| "AsyncSoftFileLease", | |
| "AsyncSoftFileLock", | |
| "AsyncStrictSoftFileLock", | |
| "AsyncUnixFileLock", | |
| "AsyncWindowsFileLock", | |
| "BaseAsyncFileLock", | |
| ] | |