codekingpro/portable-devtools

1from __future__ import annotations2 3import contextlib4import logging5import os6import time7import warnings8from abc import ABC, abstractmethod9from dataclasses import dataclass10from threading import local11from typing import TYPE_CHECKING, Any12from weakref import WeakValueDictionary13 14from ._error import Timeout15 16if TYPE_CHECKING:17    import sys18    from types import TracebackType19 20    if sys.version_info >= (3, 11):  # pragma: no cover (py311+)21        from typing import Self22    else:  # pragma: no cover (<py311)23        from typing_extensions import Self24 25 26_LOGGER = logging.getLogger("filelock")27 28 29# This is a helper class which is returned by :meth:`BaseFileLock.acquire` and wraps the lock to make sure __enter__30# is not called twice when entering the with statement. If we would simply return *self*, the lock would be acquired31# again in the *__enter__* method of the BaseFileLock, but not released again automatically. issue #37 (memory leak)32class AcquireReturnProxy:33    """A context-aware object that will release the lock file when exiting."""34 35    def __init__(self, lock: BaseFileLock) -> None:36        self.lock = lock37 38    def __enter__(self) -> BaseFileLock:39        return self.lock40 41    def __exit__(42        self,43        exc_type: type[BaseException] | None,44        exc_value: BaseException | None,45        traceback: TracebackType | None,46    ) -> None:47        self.lock.release()48 49 50@dataclass51class FileLockContext:52    """A dataclass which holds the context for a ``BaseFileLock`` object."""53 54    # The context is held in a separate class to allow optional use of thread local storage via the55    # ThreadLocalFileContext class.56 57    #: The path to the lock file.58    lock_file: str59 60    #: The default timeout value.61    timeout: float62 63    #: The mode for the lock files64    mode: int65 66    #: Whether the lock should be blocking or not67    blocking: bool68 69    #: The file descriptor for the *_lock_file* as it is returned by the os.open() function, not None when lock held70    lock_file_fd: int | None = None71 72    #: The lock counter is used for implementing the nested locking mechanism.73    lock_counter: int = 0  # When the lock is acquired is increased and the lock is only released, when this value is 074 75 76class ThreadLocalFileContext(FileLockContext, local):77    """A thread local version of the ``FileLockContext`` class."""78 79 80class BaseFileLock(ABC, contextlib.ContextDecorator):81    """Abstract base class for a file lock object."""82 83    _instances: WeakValueDictionary[str, BaseFileLock]84 85    def __new__(  # noqa: PLR091386        cls,87        lock_file: str | os.PathLike[str],88        timeout: float = -1,89        mode: int = 0o644,90        thread_local: bool = True,  # noqa: ARG003, FBT001, FBT00291        *,92        blocking: bool = True,  # noqa: ARG00393        is_singleton: bool = False,94        **kwargs: dict[str, Any],  # capture remaining kwargs for subclasses  # noqa: ARG00395    ) -> Self:96        """Create a new lock object or if specified return the singleton instance for the lock file."""97        if not is_singleton:98            return super().__new__(cls)99 100        instance = cls._instances.get(str(lock_file))101        if not instance:102            instance = super().__new__(cls)103            cls._instances[str(lock_file)] = instance104        elif timeout != instance.timeout or mode != instance.mode:105            msg = "Singleton lock instances cannot be initialized with differing arguments"106            raise ValueError(msg)107 108        return instance  # type: ignore[return-value] # https://github.com/python/mypy/issues/15322109 110    def __init_subclass__(cls, **kwargs: dict[str, Any]) -> None:111        """Setup unique state for lock subclasses."""112        super().__init_subclass__(**kwargs)113        cls._instances = WeakValueDictionary()114 115    def __init__(  # noqa: PLR0913116        self,117        lock_file: str | os.PathLike[str],118        timeout: float = -1,119        mode: int = 0o644,120        thread_local: bool = True,  # noqa: FBT001, FBT002121        *,122        blocking: bool = True,123        is_singleton: bool = False,124    ) -> None:125        """126        Create a new lock object.127 128        :param lock_file: path to the file129        :param timeout: default timeout when acquiring the lock, in seconds. It will be used as fallback value in \130            the acquire method, if no timeout value (``None``) is given. If you want to disable the timeout, set it \131            to a negative value. A timeout of 0 means that there is exactly one attempt to acquire the file lock.132        :param mode: file permissions for the lockfile133        :param thread_local: Whether this object's internal context should be thread local or not. If this is set to \134            ``False`` then the lock will be reentrant across threads.135        :param blocking: whether the lock should be blocking or not136        :param is_singleton: If this is set to ``True`` then only one instance of this class will be created \137            per lock file. This is useful if you want to use the lock object for reentrant locking without needing \138            to pass the same object around.139 140        """141        self._is_thread_local = thread_local142        self._is_singleton = is_singleton143 144        # Create the context. Note that external code should not work with the context directly and should instead use145        # properties of this class.146        kwargs: dict[str, Any] = {147            "lock_file": os.fspath(lock_file),148            "timeout": timeout,149            "mode": mode,150            "blocking": blocking,151        }152        self._context: FileLockContext = (ThreadLocalFileContext if thread_local else FileLockContext)(**kwargs)153 154    def is_thread_local(self) -> bool:155        """:return: a flag indicating if this lock is thread local or not"""156        return self._is_thread_local157 158    @property159    def is_singleton(self) -> bool:160        """:return: a flag indicating if this lock is singleton or not"""161        return self._is_singleton162 163    @property164    def lock_file(self) -> str:165        """:return: path to the lock file"""166        return self._context.lock_file167 168    @property169    def timeout(self) -> float:170        """171        :return: the default timeout value, in seconds172 173        .. versionadded:: 2.0.0174        """175        return self._context.timeout176 177    @timeout.setter178    def timeout(self, value: float | str) -> None:179        """180        Change the default timeout value.181 182        :param value: the new value, in seconds183 184        """185        self._context.timeout = float(value)186 187    @property188    def blocking(self) -> bool:189        """:return: whether the locking is blocking or not"""190        return self._context.blocking191 192    @blocking.setter193    def blocking(self, value: bool) -> None:194        """195        Change the default blocking value.196 197        :param value: the new value as bool198 199        """200        self._context.blocking = value201 202    @property203    def mode(self) -> int:204        """:return: the file permissions for the lockfile"""205        return self._context.mode206 207    @abstractmethod208    def _acquire(self) -> None:209        """If the file lock could be acquired, self._context.lock_file_fd holds the file descriptor of the lock file."""210        raise NotImplementedError211 212    @abstractmethod213    def _release(self) -> None:214        """Releases the lock and sets self._context.lock_file_fd to None."""215        raise NotImplementedError216 217    @property218    def is_locked(self) -> bool:219        """220 221        :return: A boolean indicating if the lock file is holding the lock currently.222 223        .. versionchanged:: 2.0.0224 225            This was previously a method and is now a property.226        """227        return self._context.lock_file_fd is not None228 229    @property230    def lock_counter(self) -> int:231        """:return: The number of times this lock has been acquired (but not yet released)."""232        return self._context.lock_counter233 234    def acquire(235        self,236        timeout: float | None = None,237        poll_interval: float = 0.05,238        *,239        poll_intervall: float | None = None,240        blocking: bool | None = None,241    ) -> AcquireReturnProxy:242        """243        Try to acquire the file lock.244 245        :param timeout: maximum wait time for acquiring the lock, ``None`` means use the default :attr:`~timeout` is and246         if ``timeout < 0``, there is no timeout and this method will block until the lock could be acquired247        :param poll_interval: interval of trying to acquire the lock file248        :param poll_intervall: deprecated, kept for backwards compatibility, use ``poll_interval`` instead249        :param blocking: defaults to True. If False, function will return immediately if it cannot obtain a lock on the250         first attempt. Otherwise, this method will block until the timeout expires or the lock is acquired.251        :raises Timeout: if fails to acquire lock within the timeout period252        :return: a context object that will unlock the file when the context is exited253 254        .. code-block:: python255 256            # You can use this method in the context manager (recommended)257            with lock.acquire():258                pass259 260            # Or use an equivalent try-finally construct:261            lock.acquire()262            try:263                pass264            finally:265                lock.release()266 267        .. versionchanged:: 2.0.0268 269            This method returns now a *proxy* object instead of *self*,270            so that it can be used in a with statement without side effects.271 272        """273        # Use the default timeout, if no timeout is provided.274        if timeout is None:275            timeout = self._context.timeout276 277        if blocking is None:278            blocking = self._context.blocking279 280        if poll_intervall is not None:281            msg = "use poll_interval instead of poll_intervall"282            warnings.warn(msg, DeprecationWarning, stacklevel=2)283            poll_interval = poll_intervall284 285        # Increment the number right at the beginning. We can still undo it, if something fails.286        self._context.lock_counter += 1287 288        lock_id = id(self)289        lock_filename = self.lock_file290        start_time = time.perf_counter()291        try:292            while True:293                if not self.is_locked:294                    _LOGGER.debug("Attempting to acquire lock %s on %s", lock_id, lock_filename)295                    self._acquire()296                if self.is_locked:297                    _LOGGER.debug("Lock %s acquired on %s", lock_id, lock_filename)298                    break299                if blocking is False:300                    _LOGGER.debug("Failed to immediately acquire lock %s on %s", lock_id, lock_filename)301                    raise Timeout(lock_filename)  # noqa: TRY301302                if 0 <= timeout < time.perf_counter() - start_time:303                    _LOGGER.debug("Timeout on acquiring lock %s on %s", lock_id, lock_filename)304                    raise Timeout(lock_filename)  # noqa: TRY301305                msg = "Lock %s not acquired on %s, waiting %s seconds ..."306                _LOGGER.debug(msg, lock_id, lock_filename, poll_interval)307                time.sleep(poll_interval)308        except BaseException:  # Something did go wrong, so decrement the counter.309            self._context.lock_counter = max(0, self._context.lock_counter - 1)310            raise311        return AcquireReturnProxy(lock=self)312 313    def release(self, force: bool = False) -> None:  # noqa: FBT001, FBT002314        """315        Releases the file lock. Please note, that the lock is only completely released, if the lock counter is 0.316        Also note, that the lock file itself is not automatically deleted.317 318        :param force: If true, the lock counter is ignored and the lock is released in every case/319 320        """321        if self.is_locked:322            self._context.lock_counter -= 1323 324            if self._context.lock_counter == 0 or force:325                lock_id, lock_filename = id(self), self.lock_file326 327                _LOGGER.debug("Attempting to release lock %s on %s", lock_id, lock_filename)328                self._release()329                self._context.lock_counter = 0330                _LOGGER.debug("Lock %s released on %s", lock_id, lock_filename)331 332    def __enter__(self) -> Self:333        """334        Acquire the lock.335 336        :return: the lock object337 338        """339        self.acquire()340        return self341 342    def __exit__(343        self,344        exc_type: type[BaseException] | None,345        exc_value: BaseException | None,346        traceback: TracebackType | None,347    ) -> None:348        """349        Release the lock.350 351        :param exc_type: the exception type if raised352        :param exc_value: the exception value if raised353        :param traceback: the exception traceback if raised354 355        """356        self.release()357 358    def __del__(self) -> None:359        """Called when the lock object is deleted."""360        self.release(force=True)361 362 363__all__ = [364    "AcquireReturnProxy",365    "BaseFileLock",366]367