Add backend parameter to dbm.open() and shelve.open() for explicit DBM backend selection

Author: furkanonder

Doc/library/dbm.rst

@@ -66,7 +66,7 @@ the Oracle Berkeley DB.
    The Unix file access mode of the file (default: octal ``0o666``),
    used only when the database has to be created.
 
-.. function:: open(file, flag='r', mode=0o666)
+.. function:: open(file, flag='r', mode=0o666, *, backend=None)
 
    Open a database and return the corresponding database object.
 
@@ -87,9 +87,35 @@ the Oracle Berkeley DB.
    :param int mode:
       |mode_param_doc|
 
+   :param backend:
+      The DBM backend implementation to use. If not specified, automatic
+      backend selection is used as described above. When specified, the
+      exact backend is used regardless of file existence or automatic detection.
+
+      Valid backends:
+
+      * ``'dbm.sqlite3'`` -- SQLite-based backend (if available)
+      * ``'dbm.gnu'`` -- GNU DBM backend (if available)
+      * ``'dbm.ndbm'`` -- NDBM backend (if available)
+      * ``'dbm.dumb'`` -- Pure Python backend (always available)
+
+      Specifying a backend improves compatibility when working with custom
+      serializers and ensures consistent behavior across different systems.
+
+      :raises TypeError: if not a string
+      :raises ValueError: if not a supported backend name
+      :raises ImportError: if the specified backend is not available
+   :type backend: str or None
+
    .. versionchanged:: 3.11
       *file* accepts a :term:`path-like object`.
 
+   .. versionchanged:: next
+      Added *backend* parameter for explicit DBM backend selection.
+      When specified, this overrides the automatic backend selection logic
+      and uses the exact backend requested, improving compatibility and
+      predictability across different systems.
+
 The object returned by :func:`~dbm.open` supports the same basic functionality as a
 :class:`dict`; keys and their corresponding values can be stored, retrieved, and
 deleted, and the :keyword:`in` operator and the :meth:`!keys` method are

Doc/library/shelve.rst

@@ -17,8 +17,8 @@ This includes most class instances, recursive data types, and objects containing
 lots of shared  sub-objects.  The keys are ordinary strings.
 
 
-.. function:: open(filename, flag='c', protocol=None, writeback=False, *, \
-                   serializer=None, deserializer=None)
+.. function:: open(filename, flag='c', protocol=None, writeback=False, \
+                   backend=None, *, serializer=None, deserializer=None)
 
    Open a persistent dictionary.  The filename specified is the base filename for
    the underlying database.  As a side-effect, an extension may be added to the
@@ -42,6 +42,13 @@ lots of shared  sub-objects.  The keys are ordinary strings.
    determine which accessed entries are mutable, nor which ones were actually
    mutated).
 
+   The optional *backend* parameter allows explicit selection of the underlying
+   DBM backend implementation. By default, :mod:`shelve` uses :func:`dbm.open`
+   with automatic backend selection. When *backend* is specified, that exact
+   backend will be used, which can improve compatibility when working with custom
+   serializers and ensure consistent behavior across different systems.
+   See :func:`dbm.open` for valid backend values.
+
    By default, :mod:`shelve` uses :func:`pickle.dumps` and :func:`pickle.loads`
    for serializing and deserializing. This can be changed by supplying
    *serializer* and *deserializer*, respectively.
@@ -68,6 +75,10 @@ lots of shared  sub-objects.  The keys are ordinary strings.
       Accepts custom *serializer* and *deserializer* functions in place of
       :func:`pickle.dumps` and :func:`pickle.loads`.
 
+   .. versionchanged:: next
+      Added *backend* parameter to specify which DBM backend implementation
+      to use, improving compatibility and consistency across different systems.
+
    .. note::
 
       Do not rely on the shelf being closed automatically; always call

Lib/dbm/__init__.py

@@ -50,7 +50,7 @@ class error(Exception):
     ndbm = None
 
 
-def open(file, flag='r', mode=0o666):
+def open(file, flag='r', mode=0o666, backend=None):
     """Open or create database at path given by *file*.
 
     Optional argument *flag* can be 'r' (default) for read-only access, 'w'
@@ -60,7 +60,21 @@ def open(file, flag='r', mode=0o666):
 
     Note: 'r' and 'w' fail if the database doesn't exist; 'c' creates it
     only if it doesn't exist; and 'n' always creates a new database.
+
+    Optional argument *backend* specifies which DBM backend to use.
+    Must be one of: 'dbm.sqlite3', 'dbm.gnu', 'dbm.ndbm', 'dbm.dumb'
     """
+    if backend is not None:
+        if not isinstance(backend, str):
+            raise TypeError(f"backend must be a string")
+        if backend not in _names:
+            raise ValueError(f"Unknown backend '{backend}'")
+        try:
+            mod = __import__(backend, fromlist=['open'])
+            return mod.open(file, flag, mode)
+        except ImportError as e:
+            raise ImportError(f"Backend '{backend}' is not available: {e}")
+
     global _defaultmod
     if _defaultmod is None:
         for name in _names:

Lib/shelve.py

@@ -236,11 +236,12 @@ class DbfilenameShelf(Shelf):
     See the module's __doc__ string for an overview of the interface.
     """
 
-    def __init__(self, filename, flag='c', protocol=None, writeback=False, *,
-                 serializer=None, deserializer=None):
+    def __init__(self, filename, flag='c', protocol=None, writeback=False,
+                 backend=None, *, serializer=None, deserializer=None):
         import dbm
-        Shelf.__init__(self, dbm.open(filename, flag), protocol, writeback,
-                       serializer=serializer, deserializer=deserializer)
+        Shelf.__init__(self, dbm.open(filename, flag, 0o666, backend),
+                       protocol, writeback, serializer=serializer,
+                       deserializer=deserializer)
 
     def clear(self):
         """Remove all items from the shelf."""
@@ -249,7 +250,7 @@ def clear(self):
         self.cache.clear()
         self.dict.clear()
 
-def open(filename, flag='c', protocol=None, writeback=False, *,
+def open(filename, flag='c', protocol=None, writeback=False, backend=None, *,
          serializer=None, deserializer=None):
     """Open a persistent dictionary for reading and writing.
 
@@ -263,5 +264,5 @@ def open(filename, flag='c', protocol=None, writeback=False, *,
     See the module's __doc__ string for an overview of the interface.
     """
 
-    return DbfilenameShelf(filename, flag, protocol, writeback,
+    return DbfilenameShelf(filename, flag, protocol, writeback, backend,
                            serializer=serializer, deserializer=deserializer)

Lib/test/test_dbm.py

@@ -302,12 +302,130 @@ def test_whichdb_sqlite3_existing_db(self):
             cx.close()
         self.assertEqual(self.dbm.whichdb(_fname), "dbm.sqlite3")
 
+    def setUp(self):
+        self.addCleanup(cleaunup_test_dir)
+        setup_test_dir()
+        self.dbm = import_helper.import_fresh_module('dbm')
+
 
+class DBMBackendTestCase(unittest.TestCase):
     def setUp(self):
         self.addCleanup(cleaunup_test_dir)
         setup_test_dir()
         self.dbm = import_helper.import_fresh_module('dbm')
 
+    @staticmethod
+    def get_available_backends():
+        available_backends = []
+        for backend_name in ['dbm.dumb', 'dbm.gnu', 'dbm.ndbm', 'dbm.sqlite3']:
+            try:
+                __import__(backend_name, fromlist=['open'])
+                available_backends.append(backend_name)
+            except ImportError:
+                pass
+        return available_backends
+
+    def test_backend_parameter_validation(self):
+        invalid_types = [123, [], {}, object(), dbm]
+        for invalid_backend in invalid_types:
+            with self.subTest(invalid_backend=invalid_backend):
+                with self.assertRaises(TypeError) as cm:
+                    dbm.open(_fname, backend=invalid_backend)
+                self.assertIn("backend must be a string", str(cm.exception))
+
+        invalid_names = ['invalid', 'postgres', 'gnu', 'dumb', 'sqlite3']
+        for invalid_name in invalid_names:
+            with self.subTest(invalid_name=invalid_name):
+                with self.assertRaises(ValueError) as cm:
+                    dbm.open(_fname, backend=invalid_name)
+                self.assertIn("Unknown backend", str(cm.exception))
+
+    def test_backend_parameter_functionality(self):
+        valid_backends = self.get_available_backends()
+        if len(valid_backends) < 1:
+            self.skipTest("Need at least 1 backends for this test")
+
+        for backend in valid_backends:
+            with self.subTest(backend=backend):
+                unique_fn = f"{_fname}_{backend}"
+                with dbm.open(unique_fn, 'c', backend=backend) as db:
+                    db[b'test_key'] = b'test_value'
+                    db[b'backend'] = backend.encode('ascii')
+
+                self.assertEqual(dbm.whichdb(unique_fn), backend)
+
+                with dbm.open(unique_fn, 'r', backend=backend) as db:
+                    self.assertEqual(db[b'test_key'], b'test_value')
+                    self.assertEqual(db[b'backend'], backend.encode('ascii'))
+
+    def test_backend_none_preserves_auto_selection(self):
+        with dbm.open(_fname, 'c') as db1:
+            db1[b'test'] = b'auto_selection'
+        backend1 = dbm.whichdb(_fname)
+
+        with dbm.open(_fname, 'c', backend=None) as db2:
+            db2[b'test'] = b'explicit_none'
+        backend2 = dbm.whichdb(_fname)
+
+        self.assertEqual(backend1, backend2)
+
+    def test_backend_with_different_flags(self):
+        valid_backends = self.get_available_backends()
+        if len(valid_backends) < 1:
+            self.skipTest("Need at least 1 backends for this test")
+
+        for backend in valid_backends:
+            with self.subTest(backend=backend):
+                unique_fn = f"{_fname}_{backend}"
+                with dbm.open(unique_fn, 'c', backend=backend) as db:
+                    db[b'flag_test'] = b'create'
+                self.assertEqual(dbm.whichdb(unique_fn), backend)
+
+                # Test 'w' flag (write existing)
+                with dbm.open(unique_fn, 'w', backend=backend) as db:
+                    self.assertEqual(db[b'flag_test'], b'create')
+                    db[b'flag_test'] = b'write'
+
+                # Test 'r' flag (read-only)
+                with dbm.open(unique_fn, 'r', backend=backend) as db:
+                    self.assertEqual(db[b'flag_test'], b'write')
+                    with self.assertRaises(dbm.error):
+                        db[b'readonly_test'] = b'should_fail'
+
+                # Test 'n' flag (new database)
+                new_file = f"{unique_fn}_new"
+                with dbm.open(new_file, 'n', backend=backend) as db:
+                    db[b'new_test'] = b'new_data'
+
+                self.assertEqual(dbm.whichdb(new_file), backend)
+
+    def test_backend_parameter_positioning(self):
+        valid_backends = self.get_available_backends()
+        if len(valid_backends) < 1:
+            self.skipTest("Need at least 1 backends for this test")
+
+        for backend in valid_backends:
+            with self.subTest(backend=backend):
+                unique_fn = f"{_fname}_{backend}"
+
+                # Test as keyword argument
+                file_1 = f"{unique_fn}_keyword"
+                with dbm.open(file_1, 'c', backend=backend) as db:
+                    db[b'test'] = b'keyword'
+
+                # Test as positional argument
+                file_2 = f"{unique_fn}_positional"
+                with dbm.open(file_2, 'c', 0o666, backend) as db:
+                    db[b'test'] = b'positional'
+
+                self.assertEqual(dbm.whichdb(file_1), backend)
+                self.assertEqual(dbm.whichdb(file_2), backend)
+
+    def test_backend_import_error_handling(self):
+        fake_backend = 'dbm.nonexistent'
+        with self.assertRaises(ValueError) as cm:
+            dbm.open(_fname, backend=fake_backend)
+        self.assertIn('Unknown backend', str(cm.exception))
 
 for mod in dbm_iterator():
     assert mod.__name__.startswith('dbm.')

Lib/test/test_shelve.py

@@ -400,6 +400,71 @@ def deserializer(data):
         self.assertRaises(shelve.ShelveError, shelve.BsdDbShelf, {}, **kwargs)
 
 
+class TestShelveBackend(unittest.TestCase):
+    dirname = os_helper.TESTFN
+    fn = os.path.join(os_helper.TESTFN, "shelftemp.db")
+
+    @staticmethod
+    def get_available_backends():
+        available_backends = []
+        for backend_name in ['dbm.dumb', 'dbm.gnu', 'dbm.ndbm', 'dbm.sqlite3']:
+            try:
+                __import__(backend_name, fromlist=['open'])
+                available_backends.append(backend_name)
+            except ImportError:
+                pass
+        return available_backends
+
+    def test_backend_parameter_validation(self):
+        valid_backends = self.get_available_backends()
+        if len(valid_backends) < 1:
+            self.skipTest("Need at least 1 backends for this test")
+        os.mkdir(self.dirname)
+        self.addCleanup(os_helper.rmtree, self.dirname)
+
+        for backend in valid_backends:
+            unique_fn = f"{self.fn}_{backend}"
+            with self.subTest(backend=backend):
+                with shelve.open(unique_fn, backend=backend) as shelf:
+                    shelf['test'] = 'data'
+                    self.assertEqual(shelf['test'], 'data')
+
+        invalid_types = [123, [], {}, object(), dbm]
+        for invalid_backend in invalid_types:
+            with self.subTest(invalid_backend=invalid_backend):
+                with self.assertRaises(TypeError):
+                    shelve.open(self.fn, backend=invalid_backend)
+
+        invalid_names = ['invalid', 'postgres', 'gnu', 'dumb', 'sqlite3']
+        for invalid_name in invalid_names:
+            with self.subTest(invalid_name=invalid_name):
+                with self.assertRaises(ValueError):
+                    shelve.open(self.fn, backend=invalid_name)
+
+    def test_backend_unavailable_error(self):
+        fake_backend = 'dbm.nonexistent'
+        with self.assertRaises(ValueError) as cm:
+            shelve.open(self.fn, backend=fake_backend)
+        self.assertIn('nonexistent', str(cm.exception))
+
+    def test_backend_compatibility_with_custom_serializers(self):
+        valid_backends = self.get_available_backends()
+        if len(valid_backends) < 1:
+            self.skipTest("Need at least 1 backends for this test")
+        os.mkdir(self.dirname)
+        self.addCleanup(os_helper.rmtree, self.dirname)
+
+        for backend in valid_backends:
+            for proto in range(pickle.HIGHEST_PROTOCOL + 1):
+                with self.subTest(backend=backend, protocol=proto):
+                    unique_fn = f"{self.fn}_{backend}_proto_{proto}"
+                    with shelve.open(unique_fn, backend=backend,
+                                     protocol=proto) as s:
+                        s["foo"] = "bar"
+                        self.assertEqual(s["foo"], "bar")
+                    actual_backend = dbm.whichdb(unique_fn)
+                    self.assertEqual(actual_backend, backend)
+
 class TestShelveBase:
     type2test = shelve.Shelf