Class AsyncReadWriteLock

src/filelock/_async_read_write.py:38–210

Async wrapper around :class:`ReadWriteLock` for use in ``asyncio`` applications. Because Python's :mod:`sqlite3` module has no async API, all blocking SQLite operations are dispatched to a thread pool via ``loop.run_in_executor()``. Reentrancy, upgrade/downgrade rules, and singleton be

class AsyncReadWriteLock:
    """
    Async wrapper around :class:`ReadWriteLock` for use in ``asyncio`` applications.

    Because Python's :mod:`sqlite3` module has no async API, all blocking SQLite operations are dispatched to a thread
    pool via ``loop.run_in_executor()``. Reentrancy, upgrade/downgrade rules, and singleton behavior are delegated
    to the underlying :class:`ReadWriteLock`.

    :param lock_file: path to the SQLite database file used as the lock
    :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
    :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
    :param is_singleton: if ``True``, reuse existing :class:`ReadWriteLock` instances for the same resolved path
    :param loop: event loop for ``run_in_executor``; ``None`` uses the running loop
    :param executor: executor for ``run_in_executor``. When ``None`` a dedicated single-thread executor is created
        and owned by this lock, ensuring every operation runs on the same thread (required for SQLite affinity); it
        is shut down by :meth:`close`. A caller-supplied executor is used as-is and never shut down here, so when no
        executor is passed remember to call :meth:`close` to release the owned one.

    .. versionadded:: 3.21.0

    """

    def __init__(  # noqa: PLR0913
        self,
        lock_file: str | os.PathLike[str],
        timeout: float = -1,
        *,
        blocking: bool = True,
        is_singleton: bool = True,
        loop: asyncio.AbstractEventLoop | None = None,
        executor: futures.Executor | None = None,
    ) -> None:
        self._lock = ReadWriteLock(lock_file, timeout, blocking=blocking, is_singleton=is_singleton)
        self._loop = loop
        self._owns_executor = executor is None
        self._executor = executor or ThreadPoolExecutor(max_workers=1)

    @property
    def lock_file(self) -> str:
        """The path to the lock file."""
        return self._lock.lock_file

    @property
    def timeout(self) -> float:
        """The default timeout."""
        return self._lock.timeout

    @property
    def blocking(self) -> bool:
        """Whether blocking is enabled by default."""
        return self._lock.blocking

    @property
    def loop(self) -> asyncio.AbstractEventLoop | None:
        """The event loop (or ``None`` for the running loop)."""
        return self._loop

    @property