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