(self, path, /, *, flag, mode)
| 36 | class _Database(MutableMapping): |
| 37 | |
| 38 | def __init__(self, path, /, *, flag, mode): |
| 39 | if hasattr(self, "_cx"): |
| 40 | raise error(_ERR_REINIT) |
| 41 | |
| 42 | path = os.fsdecode(path) |
| 43 | match flag: |
| 44 | case "r": |
| 45 | flag = "ro" |
| 46 | case "w": |
| 47 | flag = "rw" |
| 48 | case "c": |
| 49 | flag = "rwc" |
| 50 | Path(path).touch(mode=mode, exist_ok=True) |
| 51 | case "n": |
| 52 | flag = "rwc" |
| 53 | Path(path).unlink(missing_ok=True) |
| 54 | Path(path).touch(mode=mode) |
| 55 | case _: |
| 56 | raise ValueError("Flag must be one of 'r', 'w', 'c', or 'n', " |
| 57 | f"not {flag!r}") |
| 58 | |
| 59 | # We use the URI format when opening the database. |
| 60 | uri = _normalize_uri(path) |
| 61 | uri = f"{uri}?mode={flag}" |
| 62 | if flag == "ro": |
| 63 | # Add immutable=1 to allow read-only SQLite access even if wal/shm missing |
| 64 | uri += "&immutable=1" |
| 65 | |
| 66 | try: |
| 67 | self._cx = sqlite3.connect(uri, autocommit=True, uri=True) |
| 68 | except sqlite3.Error as exc: |
| 69 | raise error(str(exc)) |
| 70 | |
| 71 | if flag != "ro": |
| 72 | # This is an optimization only; it's ok if it fails. |
| 73 | with suppress(sqlite3.OperationalError): |
| 74 | self._cx.execute("PRAGMA journal_mode = wal") |
| 75 | |
| 76 | if flag == "rwc": |
| 77 | self._execute(BUILD_TABLE) |
| 78 | |
| 79 | def _execute(self, *args, **kwargs): |
| 80 | if not self._cx: |
nothing calls this directly
no test coverage detected