Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_read_write.py: 24%
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
1from __future__ import annotations
3import logging
4import os
5import pathlib
6import sqlite3
7import sys
8import threading
9import time
10from contextlib import contextmanager, suppress
11from typing import TYPE_CHECKING, ClassVar, Final, Literal, TypeAlias, cast
12from weakref import WeakValueDictionary
14from ._api import (
15 AcquireReturnProxy,
16 _ensure_current_process,
17 _fork_transition,
18 _raise_chained_errors,
19 _register_fork_class,
20 _register_fork_object,
21)
22from ._error import Timeout
24if TYPE_CHECKING:
25 from collections.abc import Callable, Generator
27 from _typeshed import Unused
29 if sys.version_info >= (3, 11):
30 from typing import Self
31 else:
32 from typing_extensions import Self
34_LOGGER: Final[logging.Logger] = logging.getLogger("filelock")
35_GETPID: Final[Callable[[], int]] = os.getpid
36_IS_PYPY: Final[bool] = sys.implementation.name == "pypy"
37_NEEDS_CONNECTION_ESCROW: Final[bool] = (
38 hasattr(os, "register_at_fork") and sys.implementation.name == "cpython" and sys.version_info < (3, 12)
39)
40_ConnectionParameter: TypeAlias = (
41 str | bytes | os.PathLike[str] | os.PathLike[bytes] | float | int | type[sqlite3.Connection] | None
42)
43_DatabaseIdentity: TypeAlias = tuple[int, int]
45# sqlite3_busy_timeout() accepts a C int, max 2_147_483_647 on 32-bit. Use a lower value to be safe (~23 days).
46_MAX_SQLITE_TIMEOUT_MS: Final[int] = 2_000_000_000 - 1
47_UNSAFE_FORK_EXIT_STATUS: Final[int] = 70
50class _SQLiteTransitionContext(threading.local):
51 depth: int = 0
54_SQLITE_TRANSITION_CONTEXT: Final = _SQLiteTransitionContext()
57class _ConnectionEscrow:
58 def __init__(self) -> None:
59 self._lock = threading.RLock()
60 self._functions: tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None = None
62 def functions(
63 self,
64 ) -> tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None:
65 if not _NEEDS_CONNECTION_ESCROW:
66 return None # pragma: >=3.12 cover
67 with self._lock: # pragma: <3.12 cover # pragma: needs fork
68 if self._functions is None:
69 import ctypes # ruff:ignore[import-outside-top-level] # keep optional ctypes and its audited dlsym out of ordinary imports
71 function_type = ctypes.PYFUNCTYPE(None, ctypes.py_object)
72 increment_address = ctypes.cast(ctypes.pythonapi.Py_IncRef, ctypes.c_void_p).value
73 decrement_address = ctypes.cast(ctypes.pythonapi.Py_DecRef, ctypes.c_void_p).value
74 if increment_address is None or decrement_address is None: # pragma: no cover - resolved CPython API
75 msg = "CPython reference functions have no address"
76 raise RuntimeError(msg)
77 self._functions = (
78 cast("Callable[[sqlite3.Connection], None]", function_type(increment_address)),
79 cast("Callable[[sqlite3.Connection], None]", function_type(decrement_address)),
80 )
81 return self._functions
83 def _reset_after_fork_in_child(self) -> None: # pragma: forked child
84 self._lock = threading.RLock()
87_CONNECTION_ESCROW: Final = _ConnectionEscrow()
90class _ForkedDatabaseRegistry:
91 def __init__(self) -> None:
92 self._lock = threading.RLock()
93 self._paths: set[pathlib.Path] = set()
94 self._identities: set[_DatabaseIdentity] = set()
95 self._sqlite_used = False
96 self._all_paths_poisoned = False
98 def raise_if_poisoned(self, path: pathlib.Path) -> None:
99 identity = self.identity(path)
100 with self._lock:
101 all_paths_poisoned = self._all_paths_poisoned
102 poisoned = (
103 all_paths_poisoned or path in self._paths or (identity is not None and identity in self._identities)
104 )
105 if poisoned: # pragma: needs fork
106 msg = (
107 "ReadWriteLock is unavailable in a PyPy fork child; exec or exit before using it"
108 if all_paths_poisoned
109 else f"SQLite database {path!s} was active across fork(); exec or exit before using it in the child"
110 )
111 raise RuntimeError(msg)
113 def poison_after_fork(self, path: pathlib.Path, identity: _DatabaseIdentity | None) -> None:
114 self._paths.add(path)
115 if identity is not None:
116 self._identities.add(identity)
118 def note_sqlite_use(self) -> None:
119 if _IS_PYPY:
120 with self._lock:
121 self._sqlite_used = True
123 def _reset_after_fork_in_child(self) -> None: # pragma: forked child
124 self._lock = threading.RLock()
125 self._all_paths_poisoned = self._all_paths_poisoned or (_IS_PYPY and self._sqlite_used)
126 self._sqlite_used = False
128 @staticmethod
129 def identity(path: pathlib.Path) -> _DatabaseIdentity | None:
130 try:
131 stat_result = path.stat()
132 except OSError:
133 return None
134 return stat_result.st_dev, stat_result.st_ino
137_FORKED_DATABASES: Final = _ForkedDatabaseRegistry()
140class _ForkSafeConnection(sqlite3.Connection):
141 _creator_pid: int
142 _decrement_escrow: Callable[[sqlite3.Connection], None] | None
144 def __new__(
145 cls,
146 *_args: _ConnectionParameter,
147 **_kwargs: _ConnectionParameter,
148 ) -> Self:
149 connection = super().__new__(cls)
150 connection._creator_pid = _GETPID()
151 connection._decrement_escrow = None
152 return connection
154 def close(self) -> None:
155 with _sqlite_transition():
156 if _GETPID() != self._creator_pid: # pragma: needs fork
157 return
158 with _fork_transition():
159 sqlite3.Connection.close(self)
160 if (decrement := self._decrement_escrow) is not None: # pragma: <3.12 cover # pragma: needs fork
161 self._decrement_escrow = None
162 decrement(self)
164 def acquire_escrow( # pragma: <3.12 cover # pragma: needs fork
165 self,
166 functions: tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None,
167 ) -> None:
168 # The caller only reaches here holding the escrow functions; it skips the call entirely without them.
169 if functions is not None: # pragma: no branch
170 increment, decrement = functions
171 increment(self)
172 self._decrement_escrow = decrement
174 def __del__(self) -> None:
175 with suppress(sqlite3.Error, RuntimeError):
176 self.close()
179class _ReadWriteLockMeta(type):
180 """
181 Resolve singleton instances for ``is_singleton=True`` construction.
183 This logic lives here rather than in ReadWriteLock.get_lock so ``ReadWriteLock(path)`` returns cached instances
184 without a 2-arg ``super()`` call that type checkers cannot verify.
186 """
188 _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock]
189 _instances_lock: threading.RLock
190 _instances_pid: int
191 _instances_under_construction: set[pathlib.Path]
193 def __call__(
194 cls,
195 lock_file: str | os.PathLike[str],
196 timeout: float = -1,
197 *,
198 blocking: bool = True,
199 is_singleton: bool = True,
200 ) -> ReadWriteLock:
201 _ensure_current_process()
202 if cls._instances_pid != _GETPID():
203 cls._reset_class_after_fork()
204 construction_pid = _GETPID()
205 if not is_singleton:
206 instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton)
207 if _GETPID() != construction_pid: # pragma: forked child
208 msg = "ReadWriteLock construction cannot continue after fork"
209 raise RuntimeError(msg)
210 return instance
212 normalized = pathlib.Path(lock_file).resolve()
213 with cls._instances_lock:
214 if normalized not in cls._instances:
215 if normalized in cls._instances_under_construction: # pragma: no cover - exercised in an audit callback
216 msg = f"Singleton lock construction is already active for {lock_file!s}"
217 raise RuntimeError(msg)
218 construction_registry = cls._instances_under_construction
219 construction_registry.add(normalized)
220 try:
221 instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton)
222 finally:
223 if _GETPID() == construction_pid:
224 construction_registry.discard(normalized)
225 if _GETPID() != construction_pid:
226 msg = "ReadWriteLock construction cannot continue after fork"
227 raise RuntimeError(msg)
228 cls._instances[normalized] = instance
229 else:
230 instance = cls._instances[normalized]
232 if instance.timeout != timeout or instance.blocking != blocking:
233 msg = (
234 f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking},"
235 f" cannot be changed to timeout={timeout}, blocking={blocking}"
236 )
237 raise ValueError(msg)
238 return instance
240 def _reset_class_after_fork(cls) -> None: # pragma: forked child
241 cls._instances = WeakValueDictionary()
242 cls._instances_lock = threading.RLock()
243 cls._instances_pid = _GETPID()
244 cls._instances_under_construction = set()
247class ReadWriteLock(metaclass=_ReadWriteLockMeta):
248 """
249 Cross-process read-write lock backed by SQLite.
251 Allows concurrent shared readers or a single exclusive writer. The lock is reentrant within the same mode (multiple
252 ``acquire_read`` calls nest, as do multiple ``acquire_write`` calls from the same thread), but upgrading from read
253 to write or downgrading from write to read raises :class:`RuntimeError`. Write locks are pinned to the thread that
254 acquired them.
256 By default, ``is_singleton=True``: calling ``ReadWriteLock(path)`` with the same resolved path returns the same
257 instance. The path is handed to :func:`sqlite3.connect` as given, so a ``.db`` extension is a convention rather
258 than a requirement; the filesystem must be one the active SQLite VFS supports.
260 :param lock_file: path to the SQLite database file used as the lock
261 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
262 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
263 :param is_singleton: if ``True``, reuse existing instances for the same resolved path
265 .. versionadded:: 3.21.0
267 """
269 _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock] = WeakValueDictionary()
270 _instances_lock = threading.RLock()
271 _instances_pid = _GETPID()
272 _instances_under_construction: ClassVar[set[pathlib.Path]] = set()
274 def __init_subclass__(cls) -> None:
275 super().__init_subclass__()
276 cls._instances = WeakValueDictionary()
277 cls._instances_lock = threading.RLock()
278 cls._instances_pid = _GETPID()
279 cls._instances_under_construction = set()
280 _register_fork_class(cls)
282 @classmethod
283 def get_lock(
284 cls, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True
285 ) -> ReadWriteLock:
286 """
287 Return the singleton :class:`ReadWriteLock` for *lock_file*.
289 :param lock_file: path to the SQLite database file used as the lock
290 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
291 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
293 :returns: the singleton lock instance
295 :raises ValueError: if an instance already exists for this path with different *timeout* or *blocking* values
297 """
298 return cls(lock_file, timeout, blocking=blocking)
300 def __init__(
301 self,
302 lock_file: str | os.PathLike[str],
303 timeout: float = -1,
304 *,
305 blocking: bool = True,
306 is_singleton: bool = True, # ruff:ignore[unused-method-argument] # consumed by _ReadWriteLockMeta.__call__
307 ) -> None:
308 self.lock_file = os.fspath(lock_file)
309 self._canonical_path = pathlib.Path(lock_file).resolve()
310 _FORKED_DATABASES.raise_if_poisoned(self._canonical_path)
311 self.timeout = timeout
312 self.blocking = blocking
313 self._transaction_lock = threading.Lock() # serializes the (possibly blocking) SQLite transaction work
314 self._internal_lock = threading.Lock() # protects _lock_level / _current_mode updates and rollback
315 self._lock_level = 0
316 self._current_mode: Literal["read", "write"] | None = None
317 self._write_thread_id: int | None = None
318 self._acquisition_thread_ids: set[int] = set()
319 self._con: _ForkSafeConnection | None = None
320 self._connection_transaction_released = True
321 self._connection_identity: _DatabaseIdentity | None = None
322 self._closed = False
323 self._creator_pid = _GETPID()
324 self._fork_invalidated = False
325 _register_fork_object(self)
326 with _fork_transition(), _sqlite_transition():
327 validation_connection = self._open_connection(sqlite_timeout=5.0)
328 validation_connection.close()
330 def acquire_read(self, timeout: float = -1, *, blocking: bool = True) -> AcquireReturnProxy:
331 """
332 Acquire a shared read lock.
334 If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a
335 read lock while holding a write lock raises :class:`RuntimeError` (downgrade not allowed).
337 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
338 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
340 :returns: a proxy that can be used as a context manager to release the lock
342 :raises RuntimeError: if a write lock is already held on this instance
343 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
345 """
346 return self._acquire("read", timeout, blocking=blocking)
348 def acquire_write(self, timeout: float = -1, *, blocking: bool = True) -> AcquireReturnProxy:
349 """
350 Acquire an exclusive write lock.
352 If this instance already holds a write lock from the same thread, the lock level is incremented (reentrant).
353 Attempting to acquire a write lock while holding a read lock raises :class:`RuntimeError` (upgrade not allowed).
354 Write locks are pinned to the acquiring thread: a different thread trying to re-enter also raises
355 :class:`RuntimeError`.
357 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely
358 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable
360 :returns: a proxy that can be used as a context manager to release the lock
362 :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread
363 :raises Timeout: if the lock cannot be acquired within *timeout* seconds
365 """
366 return self._acquire("write", timeout, blocking=blocking)
368 @contextmanager
369 def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
370 """
371 Context manager that acquires and releases a shared read lock.
373 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
375 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
376 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
378 """
379 if timeout is None:
380 timeout = self.timeout
381 if blocking is None:
382 blocking = self.blocking
383 self.acquire_read(timeout, blocking=blocking)
384 try:
385 yield
386 finally:
387 self.release()
389 @contextmanager
390 def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]:
391 """
392 Context manager that acquires and releases an exclusive write lock.
394 Falls back to instance defaults for *timeout* and *blocking* when ``None``.
396 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default
397 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default
399 """
400 if timeout is None:
401 timeout = self.timeout
402 if blocking is None:
403 blocking = self.blocking
404 self.acquire_write(timeout, blocking=blocking)
405 try:
406 yield
407 finally:
408 self.release()
410 def release(self, *, force: bool = False) -> None:
411 """
412 Release one level of the current lock.
414 When the lock level reaches zero the underlying SQLite transaction is rolled back, releasing the database lock.
416 :param force: if ``True``, release the lock completely regardless of the current lock level
418 :raises RuntimeError: if no lock is currently held and *force* is ``False``
420 """
421 with _fork_transition():
422 _ensure_current_process()
423 if self._inherited: # pragma: needs fork
424 return
425 self._raise_if_acquiring("release")
426 self._release(force=force, close=False)
428 def close(self) -> None:
429 """
430 Release the lock (if held) and close the underlying SQLite connection.
432 After calling this method, the lock instance is no longer usable.
434 """
435 with _fork_transition():
436 _ensure_current_process()
437 if self._inherited: # pragma: needs fork
438 return
439 self._raise_if_acquiring("close")
440 self._release(force=True, close=True)
442 def _release(self, *, force: bool, close: bool) -> None:
443 with self._transaction_lock, self._internal_lock:
444 if self._lock_level == 0:
445 if force and self._con is None:
446 if close:
447 self._closed = True
448 return
449 if not force:
450 msg = f"Cannot release a lock on {self.lock_file} (lock id: {id(self)}) that is not held"
451 raise RuntimeError(msg)
452 if not force and self._lock_level > 1:
453 self._lock_level -= 1
454 return
455 try:
456 self._finish_connection()
457 except sqlite3.Error:
458 if self._connection_transaction_released:
459 self._clear_lock_state()
460 raise
461 self._clear_lock_state()
462 if close:
463 self._closed = True
465 def _clear_lock_state(self) -> None:
466 self._lock_level = 0
467 self._current_mode = None
468 self._write_thread_id = None
470 def __del__(self) -> None:
471 if _GETPID() == getattr(self, "_creator_pid", None) and (connection := getattr(self, "_con", None)) is not None:
472 with suppress(sqlite3.Error, RuntimeError):
473 connection.close()
475 def _reset_after_fork_in_child(self) -> None: # pragma: forked child
476 if self._con is not None:
477 _FORKED_DATABASES.poison_after_fork(self._canonical_path, self._connection_identity)
478 self._con = None
479 self._connection_transaction_released = True
480 self._connection_identity = None
481 self._transaction_lock = threading.Lock()
482 self._internal_lock = threading.Lock()
483 self._clear_lock_state()
484 self._acquisition_thread_ids = set()
485 self._fork_invalidated = True
487 @property
488 def _inherited(self) -> bool:
489 return self._fork_invalidated or _GETPID() != self._creator_pid
491 def _raise_if_unusable(self) -> None:
492 _ensure_current_process()
493 if self._inherited: # pragma: needs fork
494 msg = f"ReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
495 raise RuntimeError(msg)
496 if self._closed:
497 msg = "Cannot operate on a closed database."
498 raise sqlite3.ProgrammingError(msg)
500 def _acquire(self, mode: Literal["read", "write"], timeout: float, *, blocking: bool) -> AcquireReturnProxy:
501 with _fork_transition():
502 self._raise_if_unusable()
503 operation_pid = _GETPID()
504 thread_id = threading.get_ident()
505 with self._internal_lock:
506 if self._lock_level > 0:
507 return self._validate_reentrant(mode)
508 if thread_id in self._acquisition_thread_ids: # pragma: no cover - exercised in an audit callback
509 msg = f"Cannot acquire ReadWriteLock on {self.lock_file} while acquisition is active in this thread"
510 raise RuntimeError(msg)
511 self._acquisition_thread_ids.add(thread_id)
512 try:
513 start_time = time.perf_counter()
514 self._acquire_transaction_lock(blocking=blocking, timeout=timeout)
515 try:
516 self._raise_if_unusable()
517 return self._do_acquire_inner(
518 mode,
519 timeout,
520 blocking=blocking,
521 operation_pid=operation_pid,
522 start_time=start_time,
523 )
524 finally:
525 self._transaction_lock.release()
526 finally:
527 with self._internal_lock:
528 self._acquisition_thread_ids.discard(thread_id)
530 def _do_acquire_inner(
531 self,
532 mode: Literal["read", "write"],
533 timeout: float,
534 *,
535 blocking: bool,
536 operation_pid: int,
537 start_time: float,
538 ) -> AcquireReturnProxy:
539 # Double-check: another thread may have acquired the lock while we waited on _transaction_lock.
540 with self._internal_lock:
541 if self._lock_level > 0:
542 return self._validate_reentrant(mode)
543 if self._con is not None:
544 self._finish_connection()
545 try:
546 self._open_for_acquisition(
547 timeout,
548 blocking=blocking,
549 operation_pid=operation_pid,
550 start_time=start_time,
551 )
552 self._configure_and_begin(
553 mode,
554 timeout,
555 blocking=blocking,
556 operation=(operation_pid, start_time),
557 )
558 self._raise_if_process_changed(operation_pid)
559 except BaseException as error:
560 acquisition_error: BaseException
561 if isinstance(error, sqlite3.OperationalError) and "database is locked" in str(error):
562 acquisition_error = Timeout(self.lock_file)
563 else:
564 acquisition_error = error
565 try:
566 self._finish_connection()
567 except sqlite3.Error as cleanup_error:
568 _raise_chained_errors(acquisition_error, cleanup_error)
569 if acquisition_error is not error:
570 raise acquisition_error from None
571 raise
572 with self._internal_lock:
573 self._raise_if_process_changed(operation_pid)
574 self._current_mode = mode
575 self._lock_level = 1
576 if mode == "write":
577 self._write_thread_id = threading.get_ident()
578 return AcquireReturnProxy(lock=self)
580 def _open_for_acquisition(self, timeout: float, *, blocking: bool, operation_pid: int, start_time: float) -> None:
581 with _sqlite_transition():
582 sqlite_timeout = (
583 timeout_for_sqlite(
584 timeout,
585 blocking=blocking,
586 already_waited=time.perf_counter() - start_time,
587 )
588 / 1000
589 )
590 connection = self._open_connection(sqlite_timeout=sqlite_timeout)
591 self._con, self._connection_transaction_released, self._connection_identity = (
592 connection,
593 False,
594 _FORKED_DATABASES.identity(self._canonical_path),
595 )
596 self._raise_if_process_changed(operation_pid)
598 def _configure_and_begin(
599 self,
600 mode: Literal["read", "write"],
601 timeout: float,
602 *,
603 blocking: bool,
604 operation: tuple[int, float],
605 ) -> None:
606 with _sqlite_transition():
607 operation_pid, start_time = operation
608 connection = cast("_ForkSafeConnection", self._con)
609 waited = time.perf_counter() - start_time
610 timeout_ms = timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited)
611 self._raise_if_process_changed(operation_pid)
612 connection.executescript(f"PRAGMA busy_timeout={timeout_ms}; PRAGMA journal_mode=MEMORY;").close()
613 # Use legacy journal mode (not WAL) because WAL does not block readers while a concurrent EXCLUSIVE
614 # write transaction is active, which makes read-write locking impossible without modifying table data.
615 # MEMORY is safe here since no writes happen, so a crash cannot corrupt the DB.
616 # See https://sqlite.org/lang_transaction.html#deferred_immediate_and_exclusive_transactions
617 #
618 # Recompute the remaining timeout after the blocking journal_mode pragma.
619 waited = time.perf_counter() - start_time
620 recomputed = timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited)
621 self._raise_if_process_changed(operation_pid)
622 statements = f"PRAGMA busy_timeout={recomputed}; " if recomputed != timeout_ms else ""
623 statements += "BEGIN EXCLUSIVE TRANSACTION;" if mode == "write" else "BEGIN TRANSACTION;"
624 if mode == "read":
625 # SQLite takes the SHARED lock only when a statement reads; BEGIN alone stays deferred.
626 # https://www.sqlite.org/lockingv3.html#transaction_control
627 statements += " SELECT name FROM sqlite_schema LIMIT 1;"
628 connection.executescript(statements).close()
630 def _open_connection(self, *, sqlite_timeout: float) -> _ForkSafeConnection:
631 with _sqlite_transition():
632 creator_pid = _GETPID()
633 functions = _CONNECTION_ESCROW.functions()
634 if _GETPID() != creator_pid: # pragma: forked child
635 msg = "SQLite connection construction cannot continue after fork"
636 raise RuntimeError(msg)
637 connection = _connect(
638 os.fspath(self._canonical_path),
639 factory=_ForkSafeConnection,
640 timeout=sqlite_timeout,
641 )
642 if functions is not None: # pragma: <3.12 cover # pragma: needs fork
643 connection.acquire_escrow(functions)
644 if _GETPID() != creator_pid: # pragma: forked child
645 _FORKED_DATABASES.poison_after_fork(
646 self._canonical_path,
647 _FORKED_DATABASES.identity(self._canonical_path),
648 )
649 msg = "SQLite connection construction cannot continue after fork"
650 raise RuntimeError(msg)
651 return connection
653 def _finish_connection(self) -> None:
654 with _sqlite_transition():
655 if (connection := self._con) is None:
656 return
657 rollback_error: sqlite3.Error | None = None
658 if not self._connection_transaction_released:
659 if connection.in_transaction:
660 try:
661 connection.rollback()
662 except sqlite3.Error as error:
663 if connection.in_transaction:
664 raise
665 self._connection_transaction_released = True
666 rollback_error = error
667 else:
668 self._connection_transaction_released = True
669 else:
670 self._connection_transaction_released = True
671 try:
672 connection.close()
673 except sqlite3.Error as close_error:
674 if rollback_error is not None:
675 _raise_chained_errors(rollback_error, close_error)
676 raise
677 self._con = None
678 self._connection_transaction_released = True
679 self._connection_identity = None
680 if rollback_error is not None:
681 raise rollback_error
683 def _validate_reentrant(self, mode: Literal["read", "write"]) -> AcquireReturnProxy:
684 if self._current_mode != mode:
685 opposite = "write" if mode == "read" else "read"
686 direction = "downgrade" if mode == "read" else "upgrade"
687 msg = (
688 f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): "
689 f"already holding a {opposite} lock ({direction} not allowed)"
690 )
691 raise RuntimeError(msg)
692 if mode == "write" and (cur := threading.get_ident()) != self._write_thread_id:
693 msg = (
694 f"Cannot acquire write lock on {self.lock_file} (lock id: {id(self)}) "
695 f"from thread {cur} while it is held by thread {self._write_thread_id}"
696 )
697 raise RuntimeError(msg)
698 self._lock_level += 1
699 return AcquireReturnProxy(lock=self)
701 def _acquire_transaction_lock(self, *, blocking: bool, timeout: float) -> None:
702 if not blocking:
703 acquired = self._transaction_lock.acquire(blocking=False)
704 elif timeout == -1:
705 acquired = self._transaction_lock.acquire(blocking=True)
706 else:
707 acquired = self._transaction_lock.acquire(blocking=True, timeout=timeout)
708 if not acquired:
709 raise Timeout(self.lock_file) from None
711 def _raise_if_acquiring(self, operation: Literal["acquire", "close", "release"]) -> None:
712 with self._internal_lock:
713 active_in_current_thread = threading.get_ident() in self._acquisition_thread_ids
714 if active_in_current_thread: # pragma: no cover - exercised in an audit callback
715 msg = f"Cannot {operation} ReadWriteLock on {self.lock_file} while acquisition is active in this thread"
716 raise RuntimeError(msg)
718 def _raise_if_process_changed(self, operation_pid: int) -> None:
719 if _GETPID() != operation_pid or self._inherited: # pragma: forked child
720 msg = f"ReadWriteLock on {self.lock_file} was invalidated by fork(); construct a new instance"
721 raise RuntimeError(msg)
724def _connect(database: str, *, factory: type[_ForkSafeConnection], timeout: float) -> _ForkSafeConnection:
725 _FORKED_DATABASES.note_sqlite_use()
726 return sqlite3.connect(
727 database,
728 check_same_thread=False,
729 factory=factory,
730 cached_statements=0,
731 timeout=timeout,
732 )
735@contextmanager
736def _sqlite_transition() -> Generator[None]:
737 _SQLITE_TRANSITION_CONTEXT.depth += 1
738 try:
739 yield
740 finally:
741 _SQLITE_TRANSITION_CONTEXT.depth -= 1
744def _abort_forked_sqlite_transition() -> None: # pragma: forked child
745 if _SQLITE_TRANSITION_CONTEXT.depth:
746 os._exit(_UNSAFE_FORK_EXIT_STATUS) # inherited SQLite handles cannot be used or closed safely
749def _track_sqlite_use(event: str, _args: Unused) -> None:
750 if event == "sqlite3.connect":
751 _FORKED_DATABASES.note_sqlite_use()
754def timeout_for_sqlite(timeout: float, *, blocking: bool, already_waited: float) -> int:
755 if blocking is False:
756 return 0
758 if timeout == -1:
759 return _MAX_SQLITE_TIMEOUT_MS
761 if timeout < 0:
762 msg = "timeout must be a non-negative number or -1"
763 raise ValueError(msg)
765 timeout_ms = int((max(timeout - already_waited, 0) if timeout > 0 else timeout) * 1000)
766 if timeout_ms > _MAX_SQLITE_TIMEOUT_MS or timeout_ms < 0:
767 _LOGGER.warning("timeout %s is too large for SQLite, using %s ms instead", timeout, _MAX_SQLITE_TIMEOUT_MS)
768 return _MAX_SQLITE_TIMEOUT_MS
769 return timeout_ms
772_register_fork_object(_CONNECTION_ESCROW)
773_register_fork_object(_FORKED_DATABASES)
774_register_fork_class(ReadWriteLock)
775if _IS_PYPY:
776 sys.addaudithook(_track_sqlite_use) # pragma: pypy cover
777if hasattr(os, "register_at_fork"): # pragma: needs fork
778 os.register_at_fork(after_in_child=_abort_forked_sqlite_transition)
780__all__ = [
781 "ReadWriteLock",
782 "timeout_for_sqlite",
783]