(
self,
lock_file: str | os.PathLike[str],
timeout: float = -1,
*,
blocking: bool = True,
is_singleton: bool = True, # noqa: ARG002 # consumed by _ReadWriteLockMeta.__call__
)
| 139 | return cls(lock_file, timeout, blocking=blocking) |
| 140 | |
| 141 | def __init__( |
| 142 | self, |
| 143 | lock_file: str | os.PathLike[str], |
| 144 | timeout: float = -1, |
| 145 | *, |
| 146 | blocking: bool = True, |
| 147 | is_singleton: bool = True, # noqa: ARG002 # consumed by _ReadWriteLockMeta.__call__ |
| 148 | ) -> None: |
| 149 | self.lock_file = os.fspath(lock_file) |
| 150 | self.timeout = timeout |
| 151 | self.blocking = blocking |
| 152 | self._transaction_lock = threading.Lock() # serializes the (possibly blocking) SQLite transaction work |
| 153 | self._internal_lock = threading.Lock() # protects _lock_level / _current_mode updates and rollback |
| 154 | self._lock_level = 0 |
| 155 | self._current_mode: Literal["read", "write"] | None = None |
| 156 | self._write_thread_id: int | None = None |
| 157 | self._con = sqlite3.connect(self.lock_file, check_same_thread=False) |
| 158 | with _all_connections_lock: |
| 159 | _all_connections.add(self._con) |
| 160 | |
| 161 | def _acquire_transaction_lock(self, *, blocking: bool, timeout: float) -> None: |
| 162 | if not blocking: |
nothing calls this directly
no outgoing calls
no test coverage detected