Address review from Donghee, Serhiy, and myself:

- Amend dbm docstring - Add misuse guards - Let SQLite manage the rowid - Atomic delete - Hide connection member - Remove :memory: feature - Improve URI normalisation - Fix ValueError formatting
python · erlend-aasland · Feb 14, 2024 · Jan 22, 2024 · Jan 23, 2024 · Jan 23, 2024
commit 608c22917dfffa04d2b212c4bc8c41d1451b1f0a
@@ -5,7 +5,7 @@
         import dbm
         d = dbm.open(file, 'w', 0o666)
 
-The returned object is a dbm.gnu, dbm.ndbm or dbm.dumb object, dependent on the
+The returned object is a dbm.sqlite3, dbm.gnu, dbm.ndbm or dbm.dumb object, dependent on the
 type of database being opened (determined by the whichdb function) in the case
 of an existing dbm. If the dbm does not exist and the create or new flag ('c'
 or 'n') was specified, the dbm type will be determined by the availability of

@@ -1,17 +1,18 @@
 import os
 import sqlite3
+import sys
+from pathlib import Path
 from contextlib import suppress, closing
 from collections.abc import MutableMapping
 
 BUILD_TABLE = """
   CREATE TABLE IF NOT EXISTS Dict (
-    key TEXT NOT NULL,
-    value BLOB NOT NULL,
-    PRIMARY KEY (key)
+    key TEXT UNIQUE NOT NULL,
+    value BLOB NOT NULL
   )
 """
 GET_SIZE = "SELECT COUNT (key) FROM Dict"
-LOOKUP_KEY = "SELECT value FROM Dict WHERE key = ? LIMIT 1"
+LOOKUP_KEY = "SELECT value FROM Dict WHERE key = ?"
 STORE_KV = "REPLACE INTO Dict (key, value) VALUES (?, ?)"
 DELETE_KEY = "DELETE FROM Dict WHERE key = ?"
 ITER_KEYS = "SELECT key FROM Dict"
@@ -21,17 +22,29 @@ class error(OSError):
     pass
 
 
+_ERR_CLOSED = "DBM object has already been closed"
+_ERR_REINIT = "DBM object does not support reinitialization"
+
+
 def _normalize_uri_path(path):
-    path = path.replace("?", "%3f")
-    path = path.replace("#", "%23")
-    while "//" in path:
-        path = path.replace("//", "/")
-    return path
+    path = os.fsdecode(path)
+    for char in "%", "?", "#":
+        path = path.replace(char, f"%{ord(char):02x}")
+    path = Path(path)
+    if sys.platform == "win32":
+        if path.drive:
+            if not path.is_absolute():
+                path = Path(path).absolute()
+            return Path("/", path).as_posix()
+    return path.as_posix()
 
 
 class _Database(MutableMapping):
 
-    def __init__(self, path, flag):
+    def __init__(self, path, /, flag):
+        if hasattr(self, "_cx"):
+            raise error(_ERR_REINIT)
+
         match flag:
             case "r":
                 flag = "ro"
@@ -46,26 +59,28 @@ def __init__(self, path, flag):
                 except FileNotFoundError:
                     pass
             case _:
-                raise ValueError("Flag must be one of 'r', 'w', 'c', or 'n', not",
-                                 repr(flag))
+                raise ValueError("Flag must be one of 'r', 'w', 'c', or 'n', "
+                                 f"not {flag!r}")
 
         # We use the URI format when opening the database.
-        path = os.fsdecode(path)
-        if not path or path == ":memory:":
-            uri = "file:?mode=memory"
-        else:
-            path = _normalize_uri_path(path)
-            uri = f"file:{path}?mode={flag}"
+        path = _normalize_uri_path(path)
+        uri = f"file:{path}?mode={flag}"
+
+        self._cx = sqlite3.connect(uri, autocommit=True, uri=True)
+
+        # This is an optimization only; it's ok if it fails.
+        with suppress(sqlite3.OperationalError):
+            self._cx.execute("PRAGMA journal_mode = wal")
 
-        self.cx = sqlite3.connect(uri, autocommit=True, uri=True)
-        self.cx.execute("PRAGMA journal_mode = wal")
         if flag == "rwc":
             with closing(self._execute(BUILD_TABLE)):
                 pass
 
     def _execute(self, *args, **kwargs):
+        if not self._cx:
+            raise error(_ERR_CLOSED)
         try:
-            ret = self.cx.execute(*args, **kwargs)
+            ret = self._cx.execute(*args, **kwargs)
         except sqlite3.Error as exc:
             raise error(str(exc))
         else:
@@ -88,20 +103,22 @@ def __setitem__(self, key, value):
             pass
 
     def __delitem__(self, key):
-        if key not in self:
-            raise KeyError(key)
-        with closing(self._execute(DELETE_KEY, (key,))):
-            pass
+        with closing(self._execute(DELETE_KEY, (key,))) as cu:
+            if not cu.rowcount:
+                raise KeyError(key)
 
     def __iter__(self):
-        with closing(self._execute(ITER_KEYS)) as cu:
-            for row in cu:
-                yield row[0]
+        try:
+            with closing(self._execute(ITER_KEYS)) as cu:
+                    for row in cu:
+                        yield row[0]
+        except sqlite3.Error as exc:
+            raise error(str(exc))
 
     def close(self):
-        if self.cx:
-            self.cx.close()
-            self.cx = None
+        if self._cx:
+            self._cx.close()
+            self._cx = None
 
     def keys(self):
         return list(super().keys())
@@ -113,5 +130,5 @@ def __exit__(self, *args):
         self.close()
 
 
-def open(path, flag="r", mode=None):
+def open(path, /, flag="r", mode=None):
     return _Database(path, flag)
@@ -53,6 +53,52 @@ def test_readonly_iter(self):
         self.assertEqual([k for k in self.db], ["key1", "key2"])
 
 
+class ReadWrite(_SQLiteDbmTests):
+
+    def setUp(self):
+        super().setUp()
+        self.db = dbm_sqlite3.open(self.filename, "w")
+
+    def tearDown(self):
+        self.db.close()
+        super().tearDown()
+
+    def db_content(self):
+        with closing(sqlite3.connect(self.filename)) as cx:
+            keys = [r[0] for r in cx.execute("SELECT key FROM Dict")]
+            vals = [r[0] for r in cx.execute("SELECT value FROM Dict")]
+        return keys, vals
+
+    def test_readwrite_unique_key(self):
+        self.db["key"] = "value"
+        self.db["key"] = "other"
+        keys, vals = self.db_content()
+        self.assertEqual(keys, ["key"])
+        self.assertEqual(vals, ["other"])
+
+    def test_readwrite_delete(self):
+        self.db["key"] = "value"
+        self.db["new"] = "other"
+
+        del self.db["new"]
+        keys, vals = self.db_content()
+        self.assertEqual(keys, ["key"])
+        self.assertEqual(vals, ["value"])
+
+        del self.db["key"]
+        keys, vals = self.db_content()
+        self.assertEqual(keys, [])
+        self.assertEqual(vals, [])
+
+    def test_readwrite_null_key(self):
+        with self.assertRaises(dbm_sqlite3.error):
+            self.db[None] = "value"
+
+    def test_readwrite_null_value(self):
+        with self.assertRaises(dbm_sqlite3.error):
+            self.db["key"] = None
+
+
 class Misuse(_SQLiteDbmTests):
 
     def setUp(self):
@@ -72,7 +118,8 @@ def test_misuse_double_close(self):
         self.db.close()
 
     def test_misuse_invalid_flag(self):
-        with self.assertRaises(ValueError):
+        regex = "must be.*'r'.*'w'.*'c'.*'n', not 'invalid'"
+        with self.assertRaisesRegex(ValueError, regex):
             dbm_sqlite3.open(self.filename, flag="invalid")
 
     def test_misuse_double_delete(self):
@@ -85,6 +132,39 @@ def test_misuse_invalid_key(self):
         with self.assertRaises(KeyError):
             self.db["key"]
 
+    def test_misuse_iter_close1(self):
+        self.db["1"] = 1
+        it = iter(self.db)
+        self.db.close()
+        with self.assertRaises(dbm_sqlite3.error):
+            next(it)
+
+    def test_misuse_iter_close2(self):
+        self.db["1"] = 1
+        self.db["2"] = 2
+        it = iter(self.db)
+        next(it)
+        self.db.close()
+        with self.assertRaises(dbm_sqlite3.error):
+            next(it)
+
+    def test_misuse_use_after_close(self):
+        self.db.close()
+        with self.assertRaises(dbm_sqlite3.error):
+            self.db["read"]
+        with self.assertRaises(dbm_sqlite3.error):
+            self.db["write"] = "value"
+        with self.assertRaises(dbm_sqlite3.error):
+            del self.db["del"]
+        with self.assertRaises(dbm_sqlite3.error):
+            len(self.db)
+        with self.assertRaises(dbm_sqlite3.error):
+            self.db.keys()
+
+    def test_misuse_reinit(self):
+        with self.assertRaises(dbm_sqlite3.error):
+            self.db.__init__(":memory:", flag="rw")
+
 
 class DataTypes(_SQLiteDbmTests):