codekingpro/portable-devtools

1"""An asyncio-based implementation of the file lock."""2 3from __future__ import annotations4 5import asyncio6import contextlib7import logging8import os9import time10from dataclasses import dataclass11from inspect import iscoroutinefunction12from threading import local13from typing import TYPE_CHECKING, Any, NoReturn, TypeVar14 15from ._api import _UNSET_FILE_MODE, BaseFileLock, FileLockContext, FileLockMeta16from ._error import Timeout17from ._soft import SoftFileLock18from ._unix import UnixFileLock19from ._windows import WindowsFileLock20 21if TYPE_CHECKING:22    import sys23    from collections.abc import Callable24    from concurrent import futures25    from types import TracebackType26 27    if sys.version_info >= (3, 11):  # pragma: no cover (py311+)28        from typing import Self29    else:  # pragma: no cover (<py311)30        from typing_extensions import Self31 32 33_LOGGER = logging.getLogger("filelock")34 35 36@dataclass37class AsyncFileLockContext(FileLockContext):38    """A dataclass which holds the context for a ``BaseAsyncFileLock`` object."""39 40    #: Whether run in executor41    run_in_executor: bool = True42 43    #: The executor44    executor: futures.Executor | None = None45 46    #: The loop47    loop: asyncio.AbstractEventLoop | None = None48 49 50class AsyncThreadLocalFileContext(AsyncFileLockContext, local):51    """A thread local version of the ``FileLockContext`` class."""52 53 54class AsyncAcquireReturnProxy:55    """A context-aware object that will release the lock file when exiting."""56 57    def __init__(self, lock: BaseAsyncFileLock) -> None:  # noqa: D10758        self.lock = lock59 60    async def __aenter__(self) -> BaseAsyncFileLock:  # noqa: D10561        return self.lock62 63    async def __aexit__(  # noqa: D10564        self,65        exc_type: type[BaseException] | None,66        exc_value: BaseException | None,67        traceback: TracebackType | None,68    ) -> None:69        await self.lock.release()70 71 72_AT = TypeVar("_AT", bound="BaseAsyncFileLock")73 74 75class AsyncFileLockMeta(FileLockMeta):76    def __call__(  # ty: ignore[invalid-method-override]  # noqa: PLR091377        cls: type[_AT],  # noqa: N80578        lock_file: str | os.PathLike[str],79        timeout: float = -1,80        mode: int = _UNSET_FILE_MODE,81        thread_local: bool = False,  # noqa: FBT001, FBT00282        *,83        blocking: bool = True,84        is_singleton: bool = False,85        poll_interval: float = 0.05,86        lifetime: float | None = None,87        loop: asyncio.AbstractEventLoop | None = None,88        run_in_executor: bool = True,89        executor: futures.Executor | None = None,90    ) -> _AT:91        if thread_local and run_in_executor:92            msg = "run_in_executor is not supported when thread_local is True"93            raise ValueError(msg)94        return super().__call__(95            lock_file=lock_file,96            timeout=timeout,97            mode=mode,98            thread_local=thread_local,99            blocking=blocking,100            is_singleton=is_singleton,101            poll_interval=poll_interval,102            lifetime=lifetime,103            loop=loop,104            run_in_executor=run_in_executor,105            executor=executor,106        )107 108 109class BaseAsyncFileLock(BaseFileLock, metaclass=AsyncFileLockMeta):110    """111    Base class for asynchronous file locks.112 113    .. versionadded:: 3.15.0114 115    """116 117    def __init__(  # noqa: PLR0913118        self,119        lock_file: str | os.PathLike[str],120        timeout: float = -1,121        mode: int = _UNSET_FILE_MODE,122        thread_local: bool = False,  # noqa: FBT001, FBT002123        *,124        blocking: bool = True,125        is_singleton: bool = False,126        poll_interval: float = 0.05,127        lifetime: float | None = None,128        loop: asyncio.AbstractEventLoop | None = None,129        run_in_executor: bool = True,130        executor: futures.Executor | None = None,131    ) -> None:132        """133        Create a new lock object.134 135        :param lock_file: path to the file136        :param timeout: default timeout when acquiring the lock, in seconds. It will be used as fallback value in the137            acquire method, if no timeout value (``None``) is given. If you want to disable the timeout, set it to a138            negative value. A timeout of 0 means that there is exactly one attempt to acquire the file lock.139        :param mode: file permissions for the lockfile. When not specified, the OS controls permissions via umask and140            default ACLs, preserving POSIX default ACL inheritance in shared directories.141        :param thread_local: Whether this object's internal context should be thread local or not. If this is set to142            ``False`` then the lock will be reentrant across threads.143        :param blocking: whether the lock should be blocking or not144        :param is_singleton: If this is set to ``True`` then only one instance of this class will be created per lock145            file. This is useful if you want to use the lock object for reentrant locking without needing to pass the146            same object around.147        :param poll_interval: default interval for polling the lock file, in seconds. It will be used as fallback value148            in the acquire method, if no poll_interval value (``None``) is given.149        :param lifetime: maximum time in seconds a lock can be held before it is considered expired. When set, a waiting150            process will break a lock whose file modification time is older than ``lifetime`` seconds. ``None`` (the151            default) means locks never expire.152        :param loop: The event loop to use. If not specified, the running event loop will be used.153        :param run_in_executor: If this is set to ``True`` then the lock will be acquired in an executor.154        :param executor: The executor to use. If not specified, the default executor will be used.155 156        """157        self._is_thread_local = thread_local158        self._is_singleton = is_singleton159 160        # Create the context. Note that external code should not work with the context directly and should instead use161        # properties of this class.162        kwargs: dict[str, Any] = {163            "lock_file": os.fspath(lock_file),164            "timeout": timeout,165            "mode": mode,166            "blocking": blocking,167            "poll_interval": poll_interval,168            "lifetime": lifetime,169            "loop": loop,170            "run_in_executor": run_in_executor,171            "executor": executor,172        }173        self._context: AsyncFileLockContext = (AsyncThreadLocalFileContext if thread_local else AsyncFileLockContext)(174            **kwargs175        )176 177    @property178    def run_in_executor(self) -> bool:179        """:returns: whether run in executor."""180        return self._context.run_in_executor181 182    @property183    def executor(self) -> futures.Executor | None:184        """:returns: the executor."""185        return self._context.executor186 187    @executor.setter188    def executor(self, value: futures.Executor | None) -> None:  # pragma: no cover189        """190        Change the executor.191 192        :param futures.Executor | None value: the new executor or ``None``193 194        """195        self._context.executor = value196 197    @property198    def loop(self) -> asyncio.AbstractEventLoop | None:199        """:returns: the event loop."""200        return self._context.loop201 202    async def acquire(  # ty: ignore[invalid-method-override]203        self,204        timeout: float | None = None,205        poll_interval: float | None = None,206        *,207        blocking: bool | None = None,208        cancel_check: Callable[[], bool] | None = None,209    ) -> AsyncAcquireReturnProxy:210        """211        Try to acquire the file lock.212 213        :param timeout: maximum wait time for acquiring the lock, ``None`` means use the default214            :attr:`~BaseFileLock.timeout` is and if ``timeout < 0``, there is no timeout and this method will block215            until the lock could be acquired216        :param poll_interval: interval of trying to acquire the lock file, ``None`` means use the default217            :attr:`~BaseFileLock.poll_interval`218        :param blocking: defaults to True. If False, function will return immediately if it cannot obtain a lock on the219            first attempt. Otherwise, this method will block until the timeout expires or the lock is acquired.220        :param cancel_check: a callable returning ``True`` when the acquisition should be canceled. Checked on each poll221            iteration. When triggered, raises :class:`~Timeout` just like an expired timeout.222 223        :returns: a context object that will unlock the file when the context is exited224 225        :raises Timeout: if fails to acquire lock within the timeout period226 227        .. code-block:: python228 229            # You can use this method in the context manager (recommended)230            with lock.acquire():231                pass232 233            # Or use an equivalent try-finally construct:234            lock.acquire()235            try:236                pass237            finally:238                lock.release()239 240        """241        # Use the default timeout, if no timeout is provided.242        if timeout is None:243            timeout = self._context.timeout244 245        if blocking is None:246            blocking = self._context.blocking247 248        if poll_interval is None:249            poll_interval = self._context.poll_interval250 251        # Increment the number right at the beginning. We can still undo it, if something fails.252        self._context.lock_counter += 1253 254        lock_id = id(self)255        lock_filename = self.lock_file256        start_time = time.perf_counter()257        try:258            while True:259                if not self.is_locked:260                    self._try_break_expired_lock()261                    _LOGGER.debug("Attempting to acquire lock %s on %s", lock_id, lock_filename)262                    await self._run_internal_method(self._acquire)263                if self.is_locked:264                    _LOGGER.debug("Lock %s acquired on %s", lock_id, lock_filename)265                    break266                if self._check_give_up(267                    lock_id,268                    lock_filename,269                    blocking=blocking,270                    cancel_check=cancel_check,271                    timeout=timeout,272                    start_time=start_time,273                ):274                    raise Timeout(lock_filename)  # noqa: TRY301275                msg = "Lock %s not acquired on %s, waiting %s seconds ..."276                _LOGGER.debug(msg, lock_id, lock_filename, poll_interval)277                await asyncio.sleep(poll_interval)278        except BaseException:  # Something did go wrong, so decrement the counter.279            self._context.lock_counter = max(0, self._context.lock_counter - 1)280            raise281        return AsyncAcquireReturnProxy(lock=self)282 283    async def release(self, force: bool = False) -> None:  # ty: ignore[invalid-method-override]  # noqa: FBT001, FBT002284        """285        Release the file lock. The lock is only completely released when the lock counter reaches 0. The lock file286        itself is not automatically deleted.287 288        :param force: If true, the lock counter is ignored and the lock is released in every case.289 290        """291        if self.is_locked:292            self._context.lock_counter -= 1293 294            if self._context.lock_counter == 0 or force:295                lock_id, lock_filename = id(self), self.lock_file296 297                _LOGGER.debug("Attempting to release lock %s on %s", lock_id, lock_filename)298                await self._run_internal_method(self._release)299                self._context.lock_counter = 0300                _LOGGER.debug("Lock %s released on %s", lock_id, lock_filename)301 302    async def _run_internal_method(self, method: Callable[[], Any]) -> None:303        if iscoroutinefunction(method):304            await method()305        elif self.run_in_executor:306            await asyncio.get_running_loop().run_in_executor(self.executor, method)307        else:308            method()309 310    def __enter__(self) -> NoReturn:311        """Sync context manager entry is not supported because lock acquisition is a coroutine."""312        msg = "Use `async with` — acquire/release are coroutines and cannot be awaited in a sync context manager."313        raise NotImplementedError(msg)314 315    def __exit__(316        self,317        exc_type: type[BaseException] | None,318        exc_value: BaseException | None,319        traceback: object,320    ) -> None:321        """Sync context manager exit is not supported because lock release is a coroutine."""322        msg = "Use `async with` — acquire/release are coroutines and cannot be awaited in a sync context manager."323        raise NotImplementedError(msg)324 325    async def __aenter__(self) -> Self:326        """327        Acquire the lock.328 329        :returns: the lock object330 331        """332        await self.acquire()333        return self334 335    async def __aexit__(336        self,337        exc_type: type[BaseException] | None,338        exc_value: BaseException | None,339        traceback: TracebackType | None,340    ) -> None:341        """342        Release the lock.343 344        :param exc_type: the exception type if raised345        :param exc_value: the exception value if raised346        :param traceback: the exception traceback if raised347 348        """349        await self.release()350 351    def __del__(self) -> None:352        """Release on deletion — safe to call during GC even when no event loop is running."""353        with contextlib.suppress(Exception):354            try:355                loop = asyncio.get_running_loop()356            except RuntimeError:357                # No running loop — try stored loop or create one358                loop = self._context.loop if self._context.loop and not self._context.loop.is_closed() else None359            if loop is None:360                return361            if not loop.is_running():  # pragma: no cover362                loop.run_until_complete(self.release(force=True))363            else:364                loop.create_task(self.release(force=True))365 366 367class AsyncSoftFileLock(SoftFileLock, BaseAsyncFileLock):368    """Simply watches the existence of the lock file."""369 370 371class AsyncUnixFileLock(UnixFileLock, BaseAsyncFileLock):372    """Uses the :func:`fcntl.flock` to hard lock the lock file on unix systems."""373 374 375class AsyncWindowsFileLock(WindowsFileLock, BaseAsyncFileLock):376    """Uses the :func:`msvcrt.locking` to hard lock the lock file on windows systems."""377 378 379__all__ = [380    "AsyncAcquireReturnProxy",381    "AsyncSoftFileLock",382    "AsyncUnixFileLock",383    "AsyncWindowsFileLock",384    "BaseAsyncFileLock",385]386