Class BaseFileLock

src/filelock/_api.py:587–1429

Abstract base class for a file lock object. Provides the common reentrant API and state management. Subclasses implement the locking mechanism (:class:`UnixFileLock `, :class:`WindowsFileLock `, :class:`SoftFileLock <filelock.SoftFil

class BaseFileLock(contextlib.ContextDecorator, metaclass=FileLockMeta):  # ruff:ignore[too-many-public-methods]  # public config properties
    """
    Abstract base class for a file lock object.

    Provides the common reentrant API and state management. Subclasses implement the locking mechanism
    (:class:`UnixFileLock <filelock.UnixFileLock>`, :class:`WindowsFileLock <filelock.WindowsFileLock>`,
    :class:`SoftFileLock <filelock.SoftFileLock>`).

    """

    _instances: WeakValueDictionary[str, BaseFileLock]
    _instances_lock: RLock
    _instances_under_construction: set[str]

    #: How the cross-instance deadlock message names the conflicting holder; the async subclass says "task".
    _deadlock_holder_desc: str = "FileLock instance in this thread"

    #: Whether an age-based :attr:`lifetime` lease may break this lock. Only existence locks set it (they reclaim by
    #: unlinking a pathname); native OS locks leave it ``False`` since a kernel lock cannot be revoked by file age.
    _lifetime_supported: bool = False

    #: Strict-lock and lease replacements for a backend with legacy age-based expiry.
    _lifetime_replacements: tuple[str, str] | None = None

    #: Why a backend that refuses ``lifetime`` cannot honor it, named in the warning that drops the value.
    _lifetime_unsupported_reason: str = "a native OS lock cannot be broken safely by file age"

    #: Async construction adds one metaclass frame before lifetime validation.
    _constructor_lifetime_warning_stacklevel: int = 3

    #: Whether :attr:`preserve_lock_file` may be ``True``. Native locks keep the pathname on release, so they support
    #: it; existence locks unlink their marker to release and reject it.
    _preserve_lock_file_supported: bool = True

    #: Whether an :attr:`on_acquired` hook may be set. Native locks lend the descriptor out; existence locks keep
    #: protocol state in the marker and reject it.
    _on_acquired_supported: bool = True

    #: Whether a shared instance serializes its physical acquire and release behind one gate. A backend that publishes
    #: several files per owner needs it; a single-file backend is atomic and leaves it off to skip the gate entirely.
    _serialize_transitions: bool = False

    #: Ceiling in seconds on the jittered backoff between contended acquisition retries. ``0`` keeps the fixed
    #: poll cadence; a multi-file backend sets it so contending processes desynchronize instead of livelocking.
    _poll_backoff_cap: float = 0.0

    def __init_subclass__(cls, **kwargs: _SubclassValue) -> None:
        """Give each lock subclass its own singleton registry and lock."""
        super().__init_subclass__(**kwargs)
        cls._instances = WeakValueDictionary()
        cls._instances_lock = RLock()
        cls._instances_under_construction = set()
        _register_fork_class(cls)

    @classmethod
    def _reset_class_after_fork(cls) -> None:  # pragma: forked child
        cls._instances = WeakValueDictionary()
        cls._instances_lock = RLock()