( # pragma: no cover - interpreter audit hooks run with tracing disabled
event: str, args: tuple[StrOrBytesPath, ...]
)
| 36 | |
| 37 | # Typed for the sqlite3.connect payload the hook reads; every other event returns before touching args. |
| 38 | def audit_hook( # pragma: no cover - interpreter audit hooks run with tracing disabled |
| 39 | event: str, args: tuple[StrOrBytesPath, ...] |
| 40 | ) -> None: |
| 41 | nonlocal connection_events |
| 42 | if event != "sqlite3.connect": |
| 43 | return |
| 44 | database = args[0] |
| 45 | if isinstance(database, (str, bytes)) and os.fsdecode(database) == str(lock_path): |
| 46 | connection_events += 1 |
| 47 | |
| 48 | sys.addaudithook(audit_hook) |
| 49 | lock = ReadWriteLock(lock_path, is_singleton=False) |
nothing calls this directly
no outgoing calls
no test coverage detected