Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Async wrapper around :class:`ReadWriteLock` for use with ``asyncio``."""
3from __future__ import annotations
5import asyncio
6import functools
7import os
8import sqlite3
9from concurrent.futures import ThreadPoolExecutor
10from contextlib import asynccontextmanager
11from typing import TYPE_CHECKING, ParamSpec, TypeVar
13from ._api import (
14 _append_exception_context,
15 _ensure_current_process,
16 _fork_transition,
17 _register_fork_object,
18)
19from ._async import (
20 _BackendOutcome,
21 _capture_call,
22 _drain_future,
23 _future_result,
24 _raise_cancelled_error,
25 _wait_until_done,
26)
27from ._read_write import ReadWriteLock
29if TYPE_CHECKING:
30 from collections.abc import AsyncGenerator, Callable
31 from concurrent import futures
32 from types import TracebackType
34 from ._api import AcquireReturnProxy
36_P = ParamSpec("_P")
37_R = TypeVar("_R")
40class AsyncReadWriteLock:
41 """
42 Async wrapper around :class:`ReadWriteLock` for use in ``asyncio`` applications.
44 This wrapper dispatches every blocking SQLite operation to a thread pool via ``loop.run_in_executor()`` because
45 Python's :mod:`sqlite3` module has no async API. It delegates reentrancy, upgrade/downgrade rules, and singleton
46 behavior to the underlying :class:`ReadWriteLock`.
48 :param lock_file: path to the SQLite database file used as the lock
49 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
50 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
51 :param is_singleton: if ``True``, reuse existing :class:`ReadWriteLock` instances for the same resolved path
52 :param loop: event loop for ``run_in_executor``; ``None`` uses the running loop
53 :param executor: executor for ``run_in_executor``. When ``None`` this lock creates and owns a dedicated
54 single-thread executor so every operation runs on the same thread (SQLite affinity requires this) and shuts it
55 down in :meth:`close`. This lock uses a caller-supplied executor as-is and never shuts it down, so after passing
56 no executor call :meth:`close` to release the owned one.
58 .. versionadded:: 3.21.0
60 """
62 def __init__( # ruff:ignore[too-many-arguments] # public constructor: one parameter per documented lock option
63 self,
64 lock_file: str | os.PathLike[str],
65 timeout: float = -1,
66 *,
67 blocking: bool = True,
68 is_singleton: bool = True,
69 loop: asyncio.AbstractEventLoop | None = None,
70 executor: futures.Executor | None = None,
71 ) -> None:
72 creator_pid = os.getpid()
73 self._creator_pid = creator_pid
74 self._fork_invalidated = False
75 self._closed = False
76 _register_fork_object(self)
77 with _fork_transition():
78 self._lock = ReadWriteLock(lock_file, timeout, blocking=blocking, is_singleton=is_singleton)
79 self._loop = loop
80 self._owns_executor = executor is None
81 self._executor = executor or ThreadPoolExecutor(max_workers=1)
82 if os.getpid() != creator_pid: # pragma: forked child
83 msg = "AsyncReadWriteLock construction cannot continue after fork"
84 raise RuntimeError(msg)
86 @property
87 def lock_file(self) -> str:
88 """The path to the lock file."""
89 return self._lock.lock_file
91 @property
92 def timeout(self) -> float:
93 """The default timeout."""
94 return self._lock.timeout
96 @property
97 def blocking(self) -> bool:
98 """Whether blocking is enabled by default."""
99 return self._lock.blocking
101 @property
102 def loop(self) -> asyncio.AbstractEventLoop | None:
103 """The event loop (or ``None`` for the running loop)."""
104 return self._loop
106 @property
107 def executor(self) -> futures.Executor:
108 """The executor used for ``run_in_executor`` (a dedicated single-thread one if none was supplied)."""
109 return self._executor
111 @asynccontextmanager
112 async def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> AsyncGenerator[None]:
113 """
114 Async context manager that acquires and releases a shared read lock.
116 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
118 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
119 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
121 """
122 if timeout is None:
123 timeout = self._lock.timeout
124 if blocking is None:
125 blocking = self._lock.blocking
126 await self.acquire_read(timeout, blocking=blocking)
127 body_error: BaseException | None = None
128 try:
129 yield
130 except BaseException as error:
131 body_error = error
132 raise
133 finally:
134 await self._release_in_context(body_error)
136 @asynccontextmanager
137 async def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> AsyncGenerator[None]:
138 """
139 Async context manager that acquires and releases an exclusive write lock.
141 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
143 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
144 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
146 """
147 if timeout is None:
148 timeout = self._lock.timeout
149 if blocking is None:
150 blocking = self._lock.blocking
151 await self.acquire_write(timeout, blocking=blocking)
152 body_error: BaseException | None = None
153 try:
154 yield
155 except BaseException as error:
156 body_error = error
157 raise
158 finally:
159 await self._release_in_context(body_error)
161 async def _release_in_context(self, body_error: BaseException | None) -> None:
162 try:
163 await self.release()
164 except BaseException as release_error:
165 if body_error is not None:
166 _append_exception_context(release_error, body_error)
167 raise
169 async def acquire_read(self, timeout: float = -1, *, blocking: bool = True) -> AsyncAcquireReadWriteReturnProxy:
170 """
171 Acquire a shared read lock.
173 See :meth:`ReadWriteLock.acquire_read` for full semantics.
175 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
176 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
178 :returns: a proxy that can be used as an async context manager to release the lock
180 :raises RuntimeError: if a write lock is already held on this instance
181 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
183 """
184 self._raise_if_unusable()
185 await self._run_acquire(functools.partial(self._lock.acquire_read, timeout, blocking=blocking))
186 return AsyncAcquireReadWriteReturnProxy(lock=self)
188 async def acquire_write(self, timeout: float = -1, *, blocking: bool = True) -> AsyncAcquireReadWriteReturnProxy:
189 """
190 Acquire an exclusive write lock.
192 See :meth:`ReadWriteLock.acquire_write` for full semantics.
194 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
195 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
197 :returns: a proxy that can be used as an async context manager to release the lock
199 :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread
200 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
202 """
203 self._raise_if_unusable()
204 await self._run_acquire(functools.partial(self._lock.acquire_write, timeout, blocking=blocking))
205 return AsyncAcquireReadWriteReturnProxy(lock=self)
207 async def release(self, *, force: bool = False) -> None:
208 """
209 Release one level of the current lock.
211 See :meth:`ReadWriteLock.release` for full semantics.
213 :param force: if ``True``, release the lock completely regardless of the current lock level
215 :raises RuntimeError: if no lock is currently held and *force* is ``False``
217 """
218 _ensure_current_process()
219 if self._inherited: # pragma: needs fork
220 return
221 await self._run(self._lock.release, force=force)
223 async def close(self) -> None:
224 """
225 Release the lock (if held) and close the underlying SQLite connection.
227 After calling this method, the lock instance is no longer usable.
229 """
230 _ensure_current_process()
231 if self._inherited: # pragma: needs fork
232 return
233 if self._closed:
234 return
235 close_future = self._submit(self._lock.close)
236 try:
237 await _wait_until_done(close_future)
238 except asyncio.CancelledError as cancellation:
239 try:
240 await _drain_future(close_future)
241 except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below
242 _raise_cancelled_error(cancellation, error)
243 self._closed = True
244 self._shutdown_owned_executor()
245 raise
246 _future_result(close_future)
247 self._closed = True
248 # Wait for the worker to exit rather than letting it drain in the background: a caller that forks right
249 # after closing deserves a single-threaded process, and os.fork warns about any surviving thread.
250 if self._owns_executor:
251 await asyncio.to_thread(functools.partial(self._executor.shutdown, wait=True))
253 async def _run_acquire(self, acquire: Callable[[], AcquireReturnProxy]) -> None:
254 acquire_future = self._submit(acquire)
255 try:
256 await _wait_until_done(acquire_future)
257 except asyncio.CancelledError as cancellation:
258 try:
259 await _drain_future(acquire_future)
260 except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below
261 _raise_cancelled_error(cancellation, error)
262 try:
263 await _drain_future(self._submit(self._lock.release))
264 except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below
265 _raise_cancelled_error(cancellation, error)
266 raise
267 _future_result(acquire_future)
269 async def _run(self, func: Callable[_P, _R], *args: _P.args, **kwargs: _P.kwargs) -> _R:
270 future = self._submit(func, *args, **kwargs)
271 try:
272 await _wait_until_done(future)
273 except asyncio.CancelledError as cancellation:
274 try:
275 await _drain_future(future)
276 except BaseException as error: # ruff:ignore[blind-except] # reported with the cancellation below
277 _raise_cancelled_error(cancellation, error)
278 raise
279 return _future_result(future)
281 def _submit(
282 self, func: Callable[_P, _R], *args: _P.args, **kwargs: _P.kwargs
283 ) -> asyncio.Future[_BackendOutcome[_R]]:
284 return (self._loop or asyncio.get_running_loop()).run_in_executor(
285 self._executor,
286 _capture_call,
287 functools.partial(func, *args, **kwargs),
288 )
290 def _shutdown_owned_executor(self) -> None:
291 if self._owns_executor:
292 self._executor.shutdown(wait=False)
294 @property
295 def _inherited(self) -> bool:
296 return self._fork_invalidated or os.getpid() != self._creator_pid
298 def _raise_if_unusable(self) -> None:
299 _ensure_current_process()
300 if self._inherited: # pragma: needs fork
301 msg = f"AsyncReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
302 raise RuntimeError(msg)
303 if self._closed:
304 msg = "Cannot operate on a closed database."
305 raise sqlite3.ProgrammingError(msg)
307 def _reset_after_fork_in_child(self) -> None: # pragma: forked child
308 self._fork_invalidated = True
310 def __del__(self) -> None:
311 # Safety net when close() was never called: shut down the executor we own so its worker thread does not
312 # outlive the lock. shutdown(wait=False) never blocks.
313 if os.getpid() == getattr(self, "_creator_pid", None) and getattr(self, "_owns_executor", False):
314 self._executor.shutdown(wait=False)
317class AsyncAcquireReadWriteReturnProxy:
318 """Context-aware object that releases the async read/write lock on exit."""
320 def __init__(self, lock: AsyncReadWriteLock) -> None:
321 self.lock = lock
323 async def __aenter__(self) -> AsyncReadWriteLock:
324 return self.lock
326 async def __aexit__(
327 self,
328 exc_type: type[BaseException] | None,
329 exc_value: BaseException | None,
330 traceback: TracebackType | None,
331 ) -> None:
332 await self.lock.release()
335__all__ = [
336 "AsyncAcquireReadWriteReturnProxy",
337 "AsyncReadWriteLock",
338]