| 60 | self._functions: tuple[Callable[[sqlite3.Connection], None], Callable[[sqlite3.Connection], None]] | None = None |
| 61 | |
| 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 |
| 70 | |
| 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 |
| 82 | |
| 83 | def _reset_after_fork_in_child(self) -> None: # pragma: forked child |
| 84 | self._lock = threading.RLock() |