diff --git a/mapillary_tools/history.py b/mapillary_tools/history.py
index a0cf1311..10515313 100644
--- a/mapillary_tools/history.py
+++ b/mapillary_tools/history.py
@@ -1,24 +1,17 @@
 from __future__ import annotations
 
-import contextlib
-import dbm
 import json
 import logging
+import os
+import sqlite3
 import string
 import threading
 import time
 import typing as T
+from functools import wraps
 from pathlib import Path
 
-# dbm modules are dynamically imported, so here we explicitly import dbm.sqlite3 to make sure pyinstaller include it
-# Otherwise you will see: ImportError: no dbm clone found; tried ['dbm.sqlite3', 'dbm.gnu', 'dbm.ndbm', 'dbm.dumb']
-try:
-    import dbm.sqlite3  # type: ignore
-except ImportError:
-    pass
-
-
-from . import constants, types
+from . import constants, store, types
 from .serializer.description import DescriptionJSONSerializer
 
 JSONDict = T.Dict[str, T.Union[str, int, float, None]]
@@ -85,103 +78,140 @@ def write_history(
         fp.write(json.dumps(history))
 
 
+def _retry_on_database_lock_error(fn):
+    """
+    Decorator to retry a function if it raises a sqlite3.OperationalError with
+    "database is locked" in the message.
+    """
+
+    @wraps(fn)
+    def wrapper(*args, **kwargs):
+        while True:
+            try:
+                return fn(*args, **kwargs)
+            except sqlite3.OperationalError as ex:
+                if "database is locked" in str(ex).lower():
+                    LOG.warning(f"{str(ex)}")
+                    LOG.info("Retrying in 1 second...")
+                    time.sleep(1)
+                else:
+                    raise ex
+
+    return wrapper
+
+
 class PersistentCache:
-    _lock: contextlib.nullcontext | threading.Lock
+    _lock: threading.Lock
 
     def __init__(self, file: str):
-        # SQLite3 backend supports concurrent access without a lock
-        if dbm.whichdb(file) == "dbm.sqlite3":
-            self._lock = contextlib.nullcontext()
-        else:
-            self._lock = threading.Lock()
         self._file = file
+        self._lock = threading.Lock()
 
     def get(self, key: str) -> str | None:
+        if not self._db_existed():
+            return None
+
         s = time.perf_counter()
 
-        with self._lock:
-            with dbm.open(self._file, flag="c") as db:
-                value: bytes | None = db.get(key)
+        with store.KeyValueStore(self._file, flag="r") as db:
+            try:
+                raw_payload: bytes | None = db.get(key)  # data retrieved from db[key]
+            except Exception as ex:
+                if self._table_not_found(ex):
+                    return None
+                raise ex
 
-        if value is None:
+        if raw_payload is None:
             return None
 
-        payload = self._decode(value)
+        data: JSONDict = self._decode(raw_payload)  # JSON dict decoded from db[key]
 
-        if self._is_expired(payload):
+        if self._is_expired(data):
             return None
 
-        file_handle = payload.get("file_handle")
+        cached_value = data.get("value")  # value in the JSON dict decoded from db[key]
 
         LOG.debug(
             f"Found file handle for {key} in cache ({(time.perf_counter() - s) * 1000:.0f} ms)"
         )
 
-        return T.cast(str, file_handle)
+        return T.cast(str, cached_value)
 
-    def set(self, key: str, file_handle: str, expires_in: int = 3600 * 24 * 2) -> None:
+    @_retry_on_database_lock_error
+    def set(self, key: str, value: str, expires_in: int = 3600 * 24 * 2) -> None:
         s = time.perf_counter()
 
-        payload = {
+        data = {
             "expires_at": time.time() + expires_in,
-            "file_handle": file_handle,
+            "value": value,
         }
 
-        value: bytes = json.dumps(payload).encode("utf-8")
+        payload: bytes = json.dumps(data).encode("utf-8")
 
         with self._lock:
-            with dbm.open(self._file, flag="c") as db:
-                db[key] = value
+            with store.KeyValueStore(self._file, flag="c") as db:
+                db[key] = payload
 
         LOG.debug(
             f"Cached file handle for {key} ({(time.perf_counter() - s) * 1000:.0f} ms)"
         )
 
+    @_retry_on_database_lock_error
     def clear_expired(self) -> list[str]:
-        s = time.perf_counter()
-
         expired_keys: list[str] = []
 
-        with self._lock:
-            with dbm.open(self._file, flag="c") as db:
-                if hasattr(db, "items"):
-                    items: T.Iterable[tuple[str | bytes, bytes]] = db.items()
-                else:
-                    items = ((key, db[key]) for key in db.keys())
+        s = time.perf_counter()
 
-                for key, value in items:
-                    payload = self._decode(value)
-                    if self._is_expired(payload):
+        with self._lock:
+            with store.KeyValueStore(self._file, flag="c") as db:
+                for key, raw_payload in db.items():
+                    data = self._decode(raw_payload)
+                    if self._is_expired(data):
                         del db[key]
                         expired_keys.append(T.cast(str, key))
 
-        if expired_keys:
-            LOG.debug(
-                f"Cleared {len(expired_keys)} expired entries from the cache ({(time.perf_counter() - s) * 1000:.0f} ms)"
-            )
+        LOG.debug(
+            f"Cleared {len(expired_keys)} expired entries from the cache ({(time.perf_counter() - s) * 1000:.0f} ms)"
+        )
 
         return expired_keys
 
-    def keys(self):
-        with self._lock:
-            with dbm.open(self._file, flag="c") as db:
-                return db.keys()
+    def keys(self) -> list[str]:
+        if not self._db_existed():
+            return []
 
-    def _is_expired(self, payload: JSONDict) -> bool:
-        expires_at = payload.get("expires_at")
+        try:
+            with store.KeyValueStore(self._file, flag="r") as db:
+                return [key.decode("utf-8") for key in db.keys()]
+        except Exception as ex:
+            if self._table_not_found(ex):
+                return []
+            raise ex
+
+    def _is_expired(self, data: JSONDict) -> bool:
+        expires_at = data.get("expires_at")
         if isinstance(expires_at, (int, float)):
             return expires_at is None or expires_at <= time.time()
         return False
 
-    def _decode(self, value: bytes) -> JSONDict:
+    def _decode(self, raw_payload: bytes) -> JSONDict:
         try:
-            payload = json.loads(value.decode("utf-8"))
+            data = json.loads(raw_payload.decode("utf-8"))
         except json.JSONDecodeError as ex:
             LOG.warning(f"Failed to decode cache value: {ex}")
             return {}
 
-        if not isinstance(payload, dict):
-            LOG.warning(f"Invalid cache value format: {payload}")
+        if not isinstance(data, dict):
+            LOG.warning(f"Invalid cache value format: {raw_payload!r}")
             return {}
 
-        return payload
+        return data
+
+    def _db_existed(self) -> bool:
+        return os.path.exists(self._file)
+
+    def _table_not_found(self, ex: Exception) -> bool:
+        if isinstance(ex, sqlite3.OperationalError):
+            if "no such table" in str(ex):
+                return True
+        return False
diff --git a/mapillary_tools/store.py b/mapillary_tools/store.py
new file mode 100644
index 00000000..32253665
--- /dev/null
+++ b/mapillary_tools/store.py
@@ -0,0 +1,128 @@
+"""
+This module provides a persistent key-value store based on SQLite.
+
+This implementation is mostly copied from dbm.sqlite3 in the Python standard library,
+but works for Python >= 3.9, whereas dbm.sqlite3 is only available for Python 3.13.
+
+Source: https://github.com/python/cpython/blob/3.13/Lib/dbm/sqlite3.py
+"""
+
+import os
+import sqlite3
+import sys
+from collections.abc import MutableMapping
+from contextlib import closing, suppress
+from pathlib import Path
+
+BUILD_TABLE = """
+  CREATE TABLE IF NOT EXISTS Dict (
+    key BLOB UNIQUE NOT NULL,
+    value BLOB NOT NULL
+  )
+"""
+GET_SIZE = "SELECT COUNT (key) FROM Dict"
+LOOKUP_KEY = "SELECT value FROM Dict WHERE key = CAST(? AS BLOB)"
+STORE_KV = "REPLACE INTO Dict (key, value) VALUES (CAST(? AS BLOB), CAST(? AS BLOB))"
+DELETE_KEY = "DELETE FROM Dict WHERE key = CAST(? AS BLOB)"
+ITER_KEYS = "SELECT key FROM Dict"
+
+
+def _normalize_uri(path):
+    path = Path(path)
+    uri = path.absolute().as_uri()
+    while "//" in uri:
+        uri = uri.replace("//", "/")
+    return uri
+
+
+class KeyValueStore(MutableMapping):
+    def __init__(self, path, /, *, flag="r", mode=0o666):
+        """Open a key-value database and return the object.
+
+        The 'path' parameter is the name of the database file.
+
+        The optional 'flag' parameter can be one of ...:
+            'r' (default): open an existing database for read only access
+            'w': open an existing database for read/write access
+            'c': create a database if it does not exist; open for read/write access
+            'n': always create a new, empty database; open for read/write access
+
+        The optional 'mode' parameter is the Unix file access mode of the database;
+        only used when creating a new database. Default: 0o666.
+        """
+        path = os.fsdecode(path)
+        if flag == "r":
+            flag = "ro"
+        elif flag == "w":
+            flag = "rw"
+        elif flag == "c":
+            flag = "rwc"
+            Path(path).touch(mode=mode, exist_ok=True)
+        elif flag == "n":
+            flag = "rwc"
+            Path(path).unlink(missing_ok=True)
+            Path(path).touch(mode=mode)
+        else:
+            raise ValueError(f"Flag must be one of 'r', 'w', 'c', or 'n', not {flag!r}")
+
+        # We use the URI format when opening the database.
+        uri = _normalize_uri(path)
+        uri = f"{uri}?mode={flag}"
+
+        if sys.version_info >= (3, 12):
+            # This is the preferred way, but only available in Python 3.10 and newer.
+            self._cx = sqlite3.connect(uri, autocommit=True, uri=True)
+        else:
+            self._cx = sqlite3.connect(uri, uri=True)
+
+        # This is an optimization only; it's ok if it fails.
+        with suppress(sqlite3.OperationalError):
+            self._cx.execute("PRAGMA journal_mode = wal")
+
+        if flag == "rwc":
+            self._execute(BUILD_TABLE)
+
+    def _execute(self, *args, **kwargs):
+        if sys.version_info >= (3, 12):
+            return closing(self._cx.execute(*args, **kwargs))
+        else:
+            # Use a context manager to commit the changes
+            with self._cx:
+                return closing(self._cx.execute(*args, **kwargs))
+
+    def __len__(self):
+        with self._execute(GET_SIZE) as cu:
+            row = cu.fetchone()
+        return row[0]
+
+    def __getitem__(self, key):
+        with self._execute(LOOKUP_KEY, (key,)) as cu:
+            row = cu.fetchone()
+        if not row:
+            raise KeyError(key)
+        return row[0]
+
+    def __setitem__(self, key, value):
+        self._execute(STORE_KV, (key, value))
+
+    def __delitem__(self, key):
+        with self._execute(DELETE_KEY, (key,)) as cu:
+            if not cu.rowcount:
+                raise KeyError(key)
+
+    def __iter__(self):
+        with self._execute(ITER_KEYS) as cu:
+            for row in cu:
+                yield row[0]
+
+    def close(self):
+        self._cx.close()
+
+    def keys(self):
+        return list(super().keys())
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.close()
diff --git a/mapillary_tools/uploader.py b/mapillary_tools/uploader.py
index 71a37b6a..2107d610 100644
--- a/mapillary_tools/uploader.py
+++ b/mapillary_tools/uploader.py
@@ -1311,7 +1311,7 @@ def _is_uuid(key: str) -> bool:
 
 
 def _build_upload_cache_path(upload_options: UploadOptions) -> Path:
-    # Different python/CLI versions use different cache (dbm) formats.
+    # Different python/CLI versions use different cache formats.
     # Separate them to avoid conflicts
     py_version_parts = [str(part) for part in sys.version_info[:3]]
     version = f"py_{'_'.join(py_version_parts)}_{VERSION}"
diff --git a/tests/unit/test_persistent_cache.py b/tests/unit/test_persistent_cache.py
index 32fef006..8faaf76d 100644
--- a/tests/unit/test_persistent_cache.py
+++ b/tests/unit/test_persistent_cache.py
@@ -1,25 +1,22 @@
-import dbm
+import concurrent.futures
+import multiprocessing
 import os
-import threading
+import sqlite3
 import time
+import traceback
 
 import pytest
 
 from mapillary_tools.history import PersistentCache
 
 
-# DBM backends to test with
-DBM_BACKENDS = ["dbm.sqlite3", "dbm.gnu", "dbm.ndbm", "dbm.dumb"]
-
-
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_basic_operations_with_backend(tmpdir, dbm_backend):
+def test_basic_operations_with_backend(tmpdir):
     """Test basic operations with different DBM backends.
 
     Note: This is a demonstration of pytest's parametrize feature.
     The actual PersistentCache class might not support specifying backends.
     """
-    cache_file = os.path.join(tmpdir, dbm_backend)
+    cache_file = os.path.join(tmpdir, "cache")
     # Here you would use the backend if the cache implementation supported it
     cache = PersistentCache(cache_file)
 
@@ -31,20 +28,18 @@ def test_basic_operations_with_backend(tmpdir, dbm_backend):
     # This is just a placeholder to demonstrate pytest's parametrization
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_get_set(tmpdir, dbm_backend):
+def test_get_set(tmpdir):
     """Test basic get and set operations."""
-    cache_file = os.path.join(tmpdir, f"cache_get_set_{dbm_backend}")
+    cache_file = os.path.join(tmpdir, "cache")
     cache = PersistentCache(cache_file)
     cache.set("key1", "value1")
     assert cache.get("key1") == "value1"
     assert cache.get("nonexistent_key") is None
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_expiration(tmpdir, dbm_backend):
+def test_expiration(tmpdir):
     """Test that entries expire correctly."""
-    cache_file = os.path.join(tmpdir, f"cache_expiration_{dbm_backend}")
+    cache_file = os.path.join(tmpdir, "cache")
     cache = PersistentCache(cache_file)
 
     # Set with short expiration
@@ -64,7 +59,6 @@ def test_expiration(tmpdir, dbm_backend):
     assert cache.get("long_lived") == "value"
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
 @pytest.mark.parametrize(
     "expire_time,sleep_time,should_exist",
     [
@@ -73,13 +67,9 @@ def test_expiration(tmpdir, dbm_backend):
         (5, 2, True),  # Should not expire yet
     ],
 )
-def test_parametrized_expiration(
-    tmpdir, dbm_backend, expire_time, sleep_time, should_exist
-):
+def test_parametrized_expiration(tmpdir, expire_time, sleep_time, should_exist):
     """Test expiration with different timing combinations."""
-    cache_file = os.path.join(
-        tmpdir, f"cache_param_exp_{dbm_backend}_{expire_time}_{sleep_time}"
-    )
+    cache_file = os.path.join(tmpdir, f"cache_param_exp_{expire_time}_{sleep_time}")
     cache = PersistentCache(cache_file)
 
     key = f"key_expires_in_{expire_time}_sleeps_{sleep_time}"
@@ -93,10 +83,9 @@ def test_parametrized_expiration(
         assert cache.get(key) is None
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_clear_expired(tmpdir, dbm_backend):
+def test_clear_expired(tmpdir):
     """Test clearing expired entries."""
-    cache_file = os.path.join(tmpdir, f"cache_clear_expired_{dbm_backend}")
+    cache_file = os.path.join(tmpdir, f"cache_clear_expired")
     cache = PersistentCache(cache_file)
 
     # Test 1: Single expired key
@@ -116,10 +105,9 @@ def test_clear_expired(tmpdir, dbm_backend):
     assert cache.get("not_expired") == "value2"
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_clear_expired_multiple(tmpdir, dbm_backend):
+def test_clear_expired_multiple(tmpdir):
     """Test clearing multiple expired entries."""
-    cache_file = os.path.join(tmpdir, f"cache_clear_multiple_{dbm_backend}")
+    cache_file = os.path.join(tmpdir, f"cache_clear_multiple")
     cache = PersistentCache(cache_file)
 
     # Test 2: Multiple expired keys
@@ -142,10 +130,9 @@ def test_clear_expired_multiple(tmpdir, dbm_backend):
     assert cache.get("not_expired") == "value3"
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_clear_expired_all(tmpdir, dbm_backend):
+def test_clear_expired_all(tmpdir):
     """Test clearing all expired entries."""
-    cache_file = os.path.join(tmpdir, f"cache_clear_all_{dbm_backend}")
+    cache_file = os.path.join(tmpdir, f"cache_clear_all")
     cache = PersistentCache(cache_file)
 
     # Test 3: All entries expired
@@ -164,10 +151,9 @@ def test_clear_expired_all(tmpdir, dbm_backend):
     assert b"key2" in expired_keys
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_clear_expired_none(tmpdir, dbm_backend):
+def test_clear_expired_none(tmpdir):
     """Test clearing when no entries are expired."""
-    cache_file = os.path.join(tmpdir, f"cache_clear_none_{dbm_backend}")
+    cache_file = os.path.join(tmpdir, f"cache_clear_none")
     cache = PersistentCache(cache_file)
 
     # Test 4: No entries expired
@@ -183,10 +169,9 @@ def test_clear_expired_none(tmpdir, dbm_backend):
     assert cache.get("key2") == "value2"
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_clear_expired_empty(tmpdir, dbm_backend):
+def test_clear_expired_empty(tmpdir):
     """Test clearing expired entries on an empty cache."""
-    cache_file = os.path.join(tmpdir, f"cache_clear_empty_{dbm_backend}")
+    cache_file = os.path.join(tmpdir, f"cache_clear_empty")
     cache = PersistentCache(cache_file)
 
     # Test 5: Empty cache
@@ -196,98 +181,528 @@ def test_clear_expired_empty(tmpdir, dbm_backend):
     assert len(expired_keys) == 0
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_corrupted_data(tmpdir, dbm_backend):
-    """Test handling of corrupted data."""
-    cache_file = os.path.join(tmpdir, f"cache_corrupted_{dbm_backend}")
+def test_corrupted_data(tmpdir):
+    """Test handling of corrupted data through public interface."""
+    cache_file = os.path.join(tmpdir, f"cache_corrupted")
     cache = PersistentCache(cache_file)
 
     # Set valid entry
     cache.set("key1", "value1")
 
-    # Corrupt the data by directly writing invalid JSON
-    with dbm.open(cache_file, "c") as db:
-        db["corrupted"] = b"not valid json"
-        db["corrupted_dict"] = b'"not a dict"'
-
-    # Check that corrupted entries are handled gracefully
-    assert cache.get("corrupted") is None
-    assert cache.get("corrupted_dict") is None
-
     # Valid entries should still work
     assert cache.get("key1") == "value1"
 
-    # Clear expired should not crash on corrupted entries
+    # Clear expired should not crash
     cache.clear_expired()
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_concurrency(tmpdir, dbm_backend):
-    """Test concurrent access to the cache."""
-    cache_file = os.path.join(tmpdir, f"cache_concurrency_{dbm_backend}")
+def test_keys_basic(tmpdir):
+    """Test keys() method in read mode with empty cache."""
+    cache_file = os.path.join(tmpdir, "cache_keys_empty")
     cache = PersistentCache(cache_file)
-    num_threads = 10
-    num_operations = 50
-
-    results = []  # Store assertion failures for pytest to check after threads complete
-
-    def worker(thread_id):
-        for i in range(num_operations):
-            key = f"key_{thread_id}_{i}"
-            value = f"value_{thread_id}_{i}"
-            cache.set(key, value)
-            # Occasionally read a previously written value
-            if i > 0 and i % 5 == 0:
-                prev_key = f"key_{thread_id}_{i - 1}"
-                prev_value = cache.get(prev_key)
-                if prev_value != f"value_{thread_id}_{i - 1}":
-                    results.append(
-                        f"Expected {prev_key} to be value_{thread_id}_{i - 1}, got {prev_value}"
-                    )
+    cache.set("key1", "value1")
+
+    # Test keys on non-existent cache file
+    keys = cache.keys()
+    assert keys == ["key1"]
+
+
+def test_keys_read_mode_empty_cache(tmpdir):
+    """Test keys() method in read mode with empty cache."""
+    cache_file = os.path.join(tmpdir, "cache_keys_empty")
+    cache = PersistentCache(cache_file)
+
+    # Test keys on non-existent cache file
+    keys = cache.keys()
+    assert keys == []
+
+
+def test_keys_read_mode_with_data(tmpdir):
+    """Test keys() method in read mode with existing data."""
+    cache_file = os.path.join(tmpdir, "cache_keys_data")
+    cache = PersistentCache(cache_file)
+
+    # Add some data first
+    cache.set("key1", "value1")
+    cache.set("key2", "value2")
+    cache.set("key3", "value3")
+
+    # Test keys retrieval in read mode
+    keys = cache.keys()
+    assert len(keys) == 3
+    assert "key1" in keys
+    assert "key2" in keys
+    assert "key3" in keys
+    assert set(keys) == {"key1", "key2", "key3"}
+
+
+def test_keys_read_mode_with_expired_data(tmpdir):
+    """Test keys() method in read mode includes expired entries."""
+    cache_file = os.path.join(tmpdir, "cache_keys_expired")
+    cache = PersistentCache(cache_file)
+
+    # Add data with short expiration
+    cache.set("expired_key", "value1", expires_in=1)
+    cache.set("valid_key", "value2", expires_in=10)
+
+    # Wait for expiration
+    time.sleep(1.1)
+
+    # keys() should still return expired keys (it doesn't filter by expiration)
+    keys = cache.keys()
+    assert len(keys) == 2
+    assert "expired_key" in keys
+    assert "valid_key" in keys
+
+    # But get() should return None for expired key
+    assert cache.get("expired_key") is None
+    assert cache.get("valid_key") == "value2"
 
-    threads = []
-    for i in range(num_threads):
-        t = threading.Thread(target=worker, args=(i,))
-        threads.append(t)
-        t.start()
 
-    for t in threads:
-        t.join()
+def test_keys_write_mode_concurrent_operations(tmpdir):
+    """Test keys() method during concurrent write operations."""
+    cache_file = os.path.join(tmpdir, "cache_keys_concurrent")
+    cache = PersistentCache(cache_file)
+
+    # Initial data
+    cache.set("initial_key", "initial_value")
+
+    def write_worker(worker_id):
+        """Worker function that writes data and checks keys."""
+        key = f"worker_{worker_id}_key"
+        value = f"worker_{worker_id}_value"
+        cache.set(key, value)
+
+        # Get keys after writing
+        keys = cache.keys()
+        assert key in keys
+        return keys
 
-    # Check for any failures in threads
-    assert not results, f"Thread assertions failed: {results}"
+    # Run concurrent write operations
+    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
+        futures = [executor.submit(write_worker, i) for i in range(10)]
+        # Wait for all futures to complete
+        for f in futures:
+            f.result()
 
-    # Verify all values were written correctly
-    for i in range(num_threads):
-        for j in range(num_operations):
-            key = f"key_{i}_{j}"
-            expected_value = f"value_{i}_{j}"
-            assert cache.get(key) == expected_value
+    # Final check - should have all keys
+    final_keys = cache.keys()
+    assert "initial_key" in final_keys
+    assert len(final_keys) >= 11  # initial + 10 worker keys
 
+    # Verify all worker keys are present
+    for i in range(10):
+        assert f"worker_{i}_key" in final_keys
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_decode_invalid_data(tmpdir, dbm_backend):
-    """Test _decode method with invalid data."""
-    cache_file = os.path.join(tmpdir, f"cache_decode_invalid_{dbm_backend}")
+
+def test_keys_write_mode_after_clear_expired(tmpdir):
+    """Test keys() method after clear_expired() operations."""
+    cache_file = os.path.join(tmpdir, "cache_keys_clear")
     cache = PersistentCache(cache_file)
 
-    # Test with various invalid inputs
-    result = cache._decode(b"not valid json")
-    assert result == {}
+    # Add mixed data - some that will expire, some that won't
+    cache.set("short_lived_1", "value1", expires_in=1)
+    cache.set("short_lived_2", "value2", expires_in=1)
+    cache.set("long_lived_1", "value3", expires_in=10)
+    cache.set("long_lived_2", "value4", expires_in=10)
+
+    # Verify all keys are present initially
+    initial_keys = cache.keys()
+    assert len(initial_keys) == 4
+
+    # Wait for expiration
+    time.sleep(1.1)
+
+    # Keys should still show all entries (including expired)
+    keys_before_clear = cache.keys()
+    assert len(keys_before_clear) == 4
+
+    # Clear expired entries
+    expired_keys = cache.clear_expired()
+    assert len(expired_keys) == 2
 
-    result = cache._decode(b'"string instead of dict"')
-    assert result == {}
+    # Now keys should only show non-expired entries
+    keys_after_clear = cache.keys()
+    assert len(keys_after_clear) == 2
+    assert "long_lived_1" in keys_after_clear
+    assert "long_lived_2" in keys_after_clear
+    assert "short_lived_1" not in keys_after_clear
+    assert "short_lived_2" not in keys_after_clear
 
 
-@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
-def test_is_expired(tmpdir, dbm_backend):
-    """Test _is_expired method."""
-    cache_file = os.path.join(tmpdir, f"cache_is_expired_{dbm_backend}")
+def test_keys_write_mode_large_dataset(tmpdir):
+    """Test keys() method with large dataset in write mode."""
+    cache_file = os.path.join(tmpdir, "cache_keys_large")
     cache = PersistentCache(cache_file)
 
-    # Test with various payloads
-    assert cache._is_expired({"expires_at": time.time() - 10}) is True
-    assert cache._is_expired({"expires_at": time.time() + 10}) is False
-    assert cache._is_expired({}) is False
-    assert cache._is_expired({"expires_at": "not a number"}) is False
-    assert cache._is_expired({"expires_at": None}) is False
+    # Add a large number of entries
+    num_entries = 1000
+    for i in range(num_entries):
+        cache.set(f"key_{i:04d}", f"value_{i}")
+
+    # Test keys retrieval
+    keys = cache.keys()
+    assert len(keys) == num_entries
+
+    # Verify all keys are present
+    expected_keys = {f"key_{i:04d}" for i in range(num_entries)}
+    actual_keys = set(keys)
+    assert actual_keys == expected_keys
+
+
+def test_keys_read_mode_corrupted_database(tmpdir):
+    """Test keys() method handles corrupted database gracefully."""
+    cache_file = os.path.join(tmpdir, "cache_keys_corrupted")
+    cache = PersistentCache(cache_file)
+
+    # Add some valid data first
+    cache.set("valid_key", "valid_value")
+
+    # Verify keys work with valid data
+    keys = cache.keys()
+    assert "valid_key" in keys
+
+    # The keys() method should handle database issues gracefully
+    # (specific corruption testing would require manipulating the database file)
+    # For now, we test that it doesn't crash with normal operations
+    assert isinstance(keys, list)
+    assert all(isinstance(key, str) for key in keys)
+
+
+def test_multithread_shared_cache_comprehensive(tmpdir):
+    """Test shared cache instance across multiple threads using get->set pattern.
+
+    Tests multithread scenarios using a single shared PersistentCache instance,
+    which simulates real-world usage patterns like CachedImageUploader.upload.
+    This test covers the case where values_a and values_b can intersect (overlapping keys).
+    """
+    cache_file = os.path.join(tmpdir, "cache_shared_comprehensive")
+
+    # Initialize cache once and share across all workers
+    shared_cache = PersistentCache(cache_file)
+    shared_cache.clear_expired()
+
+    num_keys = 5_000
+
+    # Generate key-value pairs for first run (overlapping patterns to ensure intersections)
+    first_dict = {f"key_{i}": f"first_value_{i}" for i in range(num_keys)}
+    assert len(first_dict) == num_keys
+
+    s = time.perf_counter()
+    # First concurrent run with get->set pattern
+    with concurrent.futures.ThreadPoolExecutor(max_workers=16) as executor:
+        list(
+            executor.map(
+                lambda kv: _shared_worker_get_set_pattern(shared_cache, [kv]),
+                first_dict.items(),
+            )
+        )
+    print(f"First run time: {(time.perf_counter() - s) * 1000:.0f} ms")
+
+    assert len(shared_cache.keys()) == len(first_dict)
+    for key in shared_cache.keys():
+        assert shared_cache.get(key) == first_dict[key]
+
+    shared_cache.clear_expired()
+
+    assert len(shared_cache.keys()) == len(first_dict)
+    for key in shared_cache.keys():
+        assert shared_cache.get(key) == first_dict[key]
+
+    # Generate key-value pairs for first run (overlapping patterns to ensure intersections)
+    second_dict = {
+        f"key_{i}": f"second_value_{i}"
+        for i in range(num_keys // 2, num_keys // 2 + num_keys)
+    }
+
+    s = time.perf_counter()
+    # First concurrent run with get->set pattern
+    with concurrent.futures.ThreadPoolExecutor(max_workers=16) as executor:
+        list(
+            executor.map(
+                lambda kv: _shared_worker_get_set_pattern(shared_cache, [kv]),
+                second_dict.items(),
+            )
+        )
+    print(f"Second run time: {(time.perf_counter() - s) * 1000:.0f} ms")
+
+    shared_cache.clear_expired()
+
+    merged_dict = {**second_dict, **first_dict}
+    assert len(merged_dict) < len(first_dict) + len(second_dict)
+
+    assert len(shared_cache.keys()) == len(merged_dict)
+    for key in shared_cache.keys():
+        assert shared_cache.get(key) == merged_dict[key]
+
+
+# Shared worker functions for concurrency tests
+def _shared_worker_get_set_pattern(cache, key_value_pairs, expires_in=36_000):
+    """Shared worker implementation: get key -> if not exist then set key=value."""
+    for key, value in key_value_pairs:
+        # Pattern: get a key -> if not exist then set key=value
+        existing_value = cache.get(key)
+        if existing_value is None:
+            cache.set(key, value, expires_in=expires_in)
+        else:
+            value = existing_value
+
+        # Verify the value was set correctly
+        retrieved_value = cache.get(key)
+        assert retrieved_value == value, (
+            f"Expected {value}, got {retrieved_value} for key {key}"
+        )
+
+
+def _multiprocess_worker_comprehensive(args):
+    """Worker function for multiprocess comprehensive test.
+
+    Each process creates its own PersistentCache instance but uses the same cache file.
+    Keys are prefixed with process_id to avoid conflicts.
+    """
+    cache_file, process_id, num_keys = args
+
+    # Create cache instance per process (but same file)
+    cache = PersistentCache(cache_file)
+
+    # Generate process-specific key-value pairs
+    process_dict = {
+        f"process_{process_id}_key_{i}": f"process_{process_id}_value_{i}"
+        for i in range(num_keys)
+    }
+
+    # Use the same get->set pattern as the original test
+    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
+        list(
+            executor.map(
+                lambda kv: _shared_worker_get_set_pattern(cache, [kv]),
+                process_dict.items(),
+            )
+        )
+
+    # Verify all process-specific keys were set correctly
+    for key, expected_value in process_dict.items():
+        actual_value = cache.get(key)
+        assert actual_value == expected_value, (
+            f"Process {process_id}: Expected {expected_value}, got {actual_value} for key {key}"
+        )
+
+    # Return process results for verification
+    return process_id, process_dict
+
+
+def test_multiprocess_shared_cache_comprehensive(tmpdir):
+    """Test shared cache file across multiple processes using get->set pattern.
+
+    This test runs the comprehensive cache test across multiple processes where:
+    - All processes use the same cache file but create their own PersistentCache instance
+    - Each process has its own keys prefixed with process_id to avoid conflicts
+    - Reuses the logic from test_multithread_shared_cache_comprehensive
+    """
+    cache_file = os.path.join(tmpdir, "cache_multiprocess_comprehensive")
+
+    # Initialize cache and clear any existing data
+    init_cache = PersistentCache(cache_file)
+    init_cache.clear_expired()
+
+    num_processes = 4
+    keys_per_process = 1000
+
+    # Prepare arguments for each process
+    process_args = [
+        (cache_file, process_id, keys_per_process)
+        for process_id in range(num_processes)
+    ]
+
+    s = time.perf_counter()
+
+    # Run multiple processes concurrently
+    with multiprocessing.Pool(processes=num_processes) as pool:
+        results = pool.map(_multiprocess_worker_comprehensive, process_args)
+
+    print(f"Multiprocess run time: {(time.perf_counter() - s) * 1000:.0f} ms")
+
+    # Verify results from all processes
+    final_cache = PersistentCache(cache_file)
+    final_cache.clear_expired()
+
+    # Collect all expected keys and values from all processes
+    all_expected_keys = {}
+    for process_id, process_dict in results:
+        all_expected_keys.update(process_dict)
+
+    # Verify total number of keys
+    final_keys = final_cache.keys()
+    assert len(final_keys) == len(all_expected_keys), (
+        f"Expected {len(all_expected_keys)} keys, got {len(final_keys)}"
+    )
+
+    # Verify all keys from all processes are present and have correct values
+    for expected_key, expected_value in all_expected_keys.items():
+        actual_value = final_cache.get(expected_key)
+        assert actual_value == expected_value, (
+            f"Expected {expected_value}, got {actual_value} for key {expected_key}"
+        )
+
+    # Verify keys are properly distributed across processes
+    for process_id in range(num_processes):
+        process_keys = [
+            key for key in final_keys if key.startswith(f"process_{process_id}_")
+        ]
+        assert len(process_keys) == keys_per_process, (
+            f"Process {process_id} should have {keys_per_process} keys, got {len(process_keys)}"
+        )
+
+    print(
+        f"Successfully verified {len(all_expected_keys)} keys across {num_processes} processes"
+    )
+
+
+def test_multithread_write_without_database_lock_errors(tmpdir):
+    cache_file = os.path.join(tmpdir, "cache_locking")
+    assert not os.path.exists(cache_file)
+
+    cache = PersistentCache(cache_file)
+
+    def _cache_set(cache, key, num_sets):
+        for _ in range(num_sets):
+            cache.set(key, "value1")
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=40) as executor:
+        futures = [
+            executor.submit(_cache_set, cache, str(key), num_sets=1)
+            for key in range(1000)
+        ]
+        r = [f.result() for f in futures]
+
+    assert 1000 == len(cache.keys())
+
+
+def _cache_set(cache_file, key, num_sets):
+    cache = PersistentCache(cache_file)
+    for i in range(num_sets):
+        cache.set(f"key_{key}_{i}", f"value_{key}_{i}")
+
+
+def test_multiprocess_write_without_database_lock_errors(tmpdir):
+    """Test no database locking errors with multiple processes accessing the same cache file."""
+
+    cache_file = os.path.join(tmpdir, "cache_locking")
+    assert not os.path.exists(cache_file)
+
+    with concurrent.futures.ProcessPoolExecutor(max_workers=40) as executor:
+        futures = [
+            executor.submit(_cache_set, cache_file, str(key), num_sets=1)
+            for key in range(10000)
+        ]
+        r = [f.result() for f in futures]
+
+    cache = PersistentCache(cache_file)
+    assert 10000 == len(cache.keys())
+
+
+def _sqlite_insert_rows(cache_file, value, num_inserts=1):
+    while True:
+        try:
+            with sqlite3.connect(cache_file) as conn:
+                conn.execute("PRAGMA journal_mode = wal")
+                conn.execute("CREATE TABLE IF NOT EXISTS cache (key TEXT, value TEXT)")
+
+            with sqlite3.connect(cache_file) as conn:
+                for _ in range(num_inserts):
+                    conn.execute(
+                        "INSERT INTO cache (key, value) VALUES (?, ?)", ("key1", value)
+                    )
+        except sqlite3.OperationalError as e:
+            traceback.print_exc()
+            if "database is locked" in str(e):
+                time.sleep(1)
+                continue
+            else:
+                raise
+        else:
+            break
+
+
+def test_multiprocess_sqlite_database_locking(tmpdir):
+    """Test database locking with multiple threads accessing the same cache file."""
+
+    cache_file = os.path.join(tmpdir, "cache_sqlite_locking")
+    assert not os.path.exists(cache_file)
+
+    num_items = 4000
+    num_inserts = 1
+
+    with concurrent.futures.ProcessPoolExecutor(max_workers=40) as executor:
+        futures = [
+            executor.submit(
+                _sqlite_insert_rows, cache_file, str(val), num_inserts=num_inserts
+            )
+            for val in range(num_items)
+        ]
+        r = [f.result() for f in futures]
+
+    with sqlite3.connect(cache_file) as conn:
+        row_count = len([row for row in conn.execute("select * from cache")])
+
+    assert row_count == num_items * num_inserts, (
+        f"Expected {num_items * num_inserts} rows, got {row_count}"
+    )
+
+    with concurrent.futures.ProcessPoolExecutor(max_workers=40) as executor:
+        futures = [
+            executor.submit(
+                _sqlite_insert_rows, cache_file, str(val), num_inserts=num_inserts
+            )
+            for val in range(num_items)
+        ]
+        r = [f.result() for f in futures]
+
+    with sqlite3.connect(cache_file) as conn:
+        row_count = len([row for row in conn.execute("select * from cache")])
+
+    assert row_count == (num_items * num_inserts) * 2, (
+        f"Expected {(num_items * num_inserts) * 2} rows, got {row_count}"
+    )
+
+
+def test_multithread_sqlite_database_locking(tmpdir):
+    """Test database locking with multiple threads accessing the same cache file."""
+
+    cache_file = os.path.join(tmpdir, "cache_sqlite_locking")
+    assert not os.path.exists(cache_file)
+
+    num_items = 4000
+    num_inserts = 1
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=40) as executor:
+        futures = [
+            executor.submit(
+                _sqlite_insert_rows, cache_file, str(val), num_inserts=num_inserts
+            )
+            for val in range(num_items)
+        ]
+        r = [f.result() for f in futures]
+
+    with sqlite3.connect(cache_file) as conn:
+        row_count = len([row for row in conn.execute("select * from cache")])
+
+    assert row_count == num_items * num_inserts, (
+        f"Expected {num_items * num_inserts} rows, got {row_count}"
+    )
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=40) as executor:
+        futures = [
+            executor.submit(
+                _sqlite_insert_rows, cache_file, str(val), num_inserts=num_inserts
+            )
+            for val in range(num_items)
+        ]
+        r = [f.result() for f in futures]
+
+    with sqlite3.connect(cache_file) as conn:
+        row_count = len([row for row in conn.execute("select * from cache")])
+
+    assert row_count == num_items * num_inserts * 2, (
+        f"Expected {num_items * num_inserts * 2} rows, got {row_count}"
+    )
diff --git a/tests/unit/test_store.py b/tests/unit/test_store.py
new file mode 100644
index 00000000..a4a68643
--- /dev/null
+++ b/tests/unit/test_store.py
@@ -0,0 +1,345 @@
+import os
+import sqlite3
+import tempfile
+
+import pytest
+
+from mapillary_tools.store import KeyValueStore
+
+
+def test_basic_dict_operations(tmpdir):
+    """Test that KeyValueStore behaves like a dict for basic operations."""
+    db_path = tmpdir.join("test.db")
+
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        # Test setting and getting - strings are stored as bytes
+        store["key1"] = "value1"
+        store["key2"] = "value2"
+
+        assert store["key1"] == b"value1"
+        assert store["key2"] == b"value2"
+
+        # Test length
+        assert len(store) == 2
+
+        # Test iteration - keys come back as bytes
+        keys = list(store)
+        assert set(keys) == {b"key1", b"key2"}
+
+        # Test keys() method
+        assert set(store.keys()) == {b"key1", b"key2"}
+
+        # Test deletion
+        del store["key1"]
+        assert len(store) == 1
+        assert "key1" not in store
+        assert "key2" in store
+
+
+def test_keyerror_on_missing_key(tmpdir):
+    """Test that KeyError is raised for missing keys."""
+    db_path = tmpdir.join("test.db")
+
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        with pytest.raises(KeyError):
+            _ = store["nonexistent"]
+
+        with pytest.raises(KeyError):
+            del store["nonexistent"]
+
+
+def test_flag_modes(tmpdir):
+    """Test different flag modes."""
+    db_path = tmpdir.join("test.db")
+
+    # Test 'n' flag - creates new database
+    with KeyValueStore(str(db_path), flag="n", mode=0o666) as store:
+        store["key"] = "value"
+        assert store["key"] == b"value"
+
+    # Test 'c' flag - opens existing or creates new
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        assert store["key"] == b"value"  # Should still exist
+        store["key2"] = "value2"
+
+    # Test 'w' flag - opens existing for read/write
+    with KeyValueStore(str(db_path), flag="w", mode=0o666) as store:
+        assert store["key"] == b"value"
+        assert store["key2"] == b"value2"
+        store["key3"] = "value3"
+
+    # Test 'r' flag - read-only mode
+    with KeyValueStore(str(db_path), flag="r", mode=0o666) as store:
+        # Getter should work in readonly mode
+        assert store["key"] == b"value"
+        assert store["key2"] == b"value2"
+        assert store["key3"] == b"value3"
+        assert len(store) == 3
+
+        # Iteration should work in readonly mode
+        keys = set(store.keys())
+        assert keys == {b"key", b"key2", b"key3"}
+
+        # But setter should fail
+        with pytest.raises(sqlite3.OperationalError):
+            store["new_key"] = "new_value"
+
+
+def test_readonly_mode_comprehensive(tmpdir):
+    """Test comprehensive readonly mode functionality."""
+    db_path = tmpdir.join("readonly_test.db")
+
+    # First, create and populate the database
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        store["test_key"] = "test_value"
+        store["another_key"] = "another_value"
+        store[b"binary_key"] = b"binary_value"
+        store[123] = "numeric_key_value"
+
+    # Now open in readonly mode and test all read operations
+    with KeyValueStore(str(db_path), flag="r", mode=0o666) as readonly_store:
+        # Test basic getitem - strings come back as bytes
+        assert readonly_store["test_key"] == b"test_value"
+        assert readonly_store["another_key"] == b"another_value"
+        assert readonly_store[b"binary_key"] == b"binary_value"
+        assert readonly_store[123] == b"numeric_key_value"
+
+        # Test len
+        assert len(readonly_store) == 4
+
+        # Test iteration - keys come back as bytes
+        all_keys = set(readonly_store)
+        assert all_keys == {b"test_key", b"another_key", b"binary_key", b"123"}
+
+        # Test keys() method
+        assert set(readonly_store.keys()) == all_keys
+
+        # Test containment
+        assert "test_key" in readonly_store
+        assert "nonexistent" not in readonly_store
+
+        # Test that write operations fail
+        with pytest.raises(
+            sqlite3.OperationalError, match="attempt to write a readonly database"
+        ):
+            readonly_store["new_key"] = "should_fail"
+
+        with pytest.raises(
+            sqlite3.OperationalError, match="attempt to write a readonly database"
+        ):
+            del readonly_store["test_key"]
+
+
+def test_invalid_flag():
+    """Test that invalid flags raise ValueError."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = os.path.join(tmpdir, "test.db")
+
+        with pytest.raises(ValueError, match="Flag must be one of"):
+            KeyValueStore(db_path, flag="x", mode=0o666)
+
+
+def test_context_manager(tmpdir):
+    """Test context manager functionality."""
+    db_path = tmpdir.join("context_test.db")
+
+    # Test normal context manager usage
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        store["key"] = "value"
+        assert store["key"] == b"value"
+
+    # After exiting context, store should be closed
+    with pytest.raises(
+        sqlite3.ProgrammingError, match="Cannot operate on a closed database"
+    ):
+        _ = store["key"]
+
+
+def test_manual_close(tmpdir):
+    """Test manual close functionality."""
+    db_path = tmpdir.join("close_test.db")
+
+    store = KeyValueStore(str(db_path), flag="c", mode=0o666)
+    store["key"] = "value"
+    assert store["key"] == b"value"
+
+    store.close()
+
+    # After closing, operations should fail
+    with pytest.raises(
+        sqlite3.ProgrammingError, match="Cannot operate on a closed database"
+    ):
+        _ = store["key"]
+
+    with pytest.raises(
+        sqlite3.ProgrammingError, match="Cannot operate on a closed database"
+    ):
+        store["new_key"] = "new_value"
+
+
+def test_open_function(tmpdir):
+    """Test the open() function."""
+    db_path = tmpdir.join("open_test.db")
+
+    # Test default parameters
+    with KeyValueStore(str(db_path), flag="c") as store:
+        store["key"] = "value"
+        assert store["key"] == b"value"
+
+    # Test with explicit parameters
+    with KeyValueStore(str(db_path), flag="w", mode=0o644) as store:
+        assert store["key"] == b"value"
+        store["key2"] = "value2"
+
+    # Test readonly
+    with KeyValueStore(str(db_path), flag="r") as store:
+        assert store["key"] == b"value"
+        assert store["key2"] == b"value2"
+
+
+def test_binary_and_various_key_types(tmpdir):
+    """Test that the store can handle various key and value types."""
+    db_path = tmpdir.join("types_test.db")
+
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        # String keys and values
+        store["string_key"] = "string_value"
+
+        # Binary keys and values
+        store[b"binary_key"] = b"binary_value"
+
+        # Numeric keys
+        store[123] = "numeric_key"
+        store["numeric_value"] = 456
+
+        # Mixed types - use different key names to avoid collision
+        store[b"mixed_binary"] = "string_value_for_binary_key"
+        store["mixed_string"] = b"binary_value_for_string_key"
+
+        # Verify all work - strings come back as bytes
+        assert store["string_key"] == b"string_value"
+        assert store[b"binary_key"] == b"binary_value"
+        assert store[123] == b"numeric_key"
+        assert store["numeric_value"] == b"456"
+        assert store[b"mixed_binary"] == b"string_value_for_binary_key"
+        assert store["mixed_string"] == b"binary_value_for_string_key"
+
+
+def test_overwrite_behavior(tmpdir):
+    """Test that values can be overwritten."""
+    db_path = tmpdir.join("overwrite_test.db")
+
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        store["key"] = "original_value"
+        assert store["key"] == b"original_value"
+
+        store["key"] = "new_value"
+        assert store["key"] == b"new_value"
+
+        # Length should still be 1
+        assert len(store) == 1
+
+
+def test_empty_store(tmpdir):
+    """Test behavior with empty store."""
+    db_path = tmpdir.join("empty_test.db")
+
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        assert len(store) == 0
+        assert list(store) == []
+        assert store.keys() == []
+
+        with pytest.raises(KeyError):
+            _ = store["any_key"]
+
+
+def test_new_flag_overwrites_existing(tmpdir):
+    """Test that 'n' flag creates a new database, overwriting existing."""
+    db_path = tmpdir.join("new_flag_test.db")
+
+    # Create initial database
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        store["key"] = "original"
+
+    # Open with 'n' flag should create new database
+    with KeyValueStore(str(db_path), flag="n", mode=0o666) as store:
+        assert len(store) == 0  # Should be empty
+        store["key"] = "new"
+
+    # Verify the old data is gone
+    with KeyValueStore(str(db_path), flag="r", mode=0o666) as store:
+        assert store["key"] == b"new"
+        assert len(store) == 1
+
+
+def test_persistence_across_sessions(tmpdir):
+    """Test that data persists across different sessions."""
+    db_path = tmpdir.join("persistence_test.db")
+
+    # First session
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        store["persistent_key"] = "persistent_value"
+        store["another_key"] = "another_value"
+
+    # Second session
+    with KeyValueStore(str(db_path), flag="r", mode=0o666) as store:
+        assert store["persistent_key"] == b"persistent_value"
+        assert store["another_key"] == b"another_value"
+        assert len(store) == 2
+
+
+def test_dict_like_membership(tmpdir):
+    """Test membership operations work like dict."""
+    db_path = tmpdir.join("membership_test.db")
+
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        store["exists"] = "value"
+
+        assert "exists" in store
+        assert "does_not_exist" not in store
+
+        # Test with different key types
+        store[123] = "numeric"
+        store[b"binary"] = "binary_value"
+
+        assert 123 in store
+        assert b"binary" in store
+        assert 456 not in store
+        assert b"not_there" not in store
+
+
+def test_readonly_mode_nonexistent_file(tmpdir):
+    """Test readonly mode with a non-existent file."""
+    db_path = tmpdir.join("nonexistent.db")
+
+    # Try to open a non-existent file in readonly mode
+    # This should raise an error since the file doesn't exist
+    with pytest.raises(sqlite3.OperationalError, match="unable to open database file"):
+        KeyValueStore(str(db_path), flag="r", mode=0o666)
+
+
+def test_readonly_mode_empty_database(tmpdir):
+    """Test readonly mode with an empty database."""
+    db_path = tmpdir.join("empty_readonly.db")
+
+    # First create an empty database
+    with KeyValueStore(str(db_path), flag="c", mode=0o666) as store:
+        pass  # Create empty database
+
+    # Now open it in readonly mode
+    with KeyValueStore(str(db_path), flag="r", mode=0o666) as readonly_store:
+        # Getting a non-existent value should raise KeyError
+        with pytest.raises(KeyError):
+            _ = readonly_store["nonexistent_key"]
+
+        # Length should be 0
+        assert len(readonly_store) == 0
+
+        # Keys should be empty
+        assert list(readonly_store.keys()) == []
+
+        # Iteration should be empty
+        assert list(readonly_store) == []
+
+        # Membership test should return False
+        assert "any_key" not in readonly_store