(
self,
lock_file: str | os.PathLike[str],
timeout: float = -1,
*,
blocking: bool = True,
is_singleton: bool = True, # ruff:ignore[unused-method-argument] # consumed by _ReadWriteLockMeta.__call__
)
| 298 | return cls(lock_file, timeout, blocking=blocking) |
| 299 | |
| 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() |
| 329 | |
| 330 | def acquire_read(self, timeout: float = -1, *, blocking: bool = True) -> AcquireReturnProxy: |
| 331 | """ |
nothing calls this directly
no test coverage detected