checkpoint api feature support to align with tdb v8.3.2+; correct ver…

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "tidesdb"
-version = "0.9.0"
+version = "0.9.1"
 description = "Official Python bindings for TidesDB - A high-performance embedded key-value storage engine"
 readme = "README.md"
 requires-python = ">=3.10"

src/tidesdb/__init__.py

@@ -27,7 +27,7 @@
     COMPARATOR_FUNC,
 )
 
-__version__ = "7.3.1"
+__version__ = "0.9.1"
 __all__ = [
     "TidesDB",
     "Transaction",

src/tidesdb/tidesdb.py

@@ -374,6 +374,9 @@ class _CCacheStats(Structure):
 _lib.tidesdb_backup.argtypes = [c_void_p, c_char_p]
 _lib.tidesdb_backup.restype = c_int
 
+_lib.tidesdb_checkpoint.argtypes = [c_void_p, c_char_p]
+_lib.tidesdb_checkpoint.restype = c_int
+
 _lib.tidesdb_rename_column_family.argtypes = [c_void_p, c_char_p, c_char_p]
 _lib.tidesdb_rename_column_family.restype = c_int
 
@@ -1300,6 +1303,27 @@ def backup(self, backup_dir: str) -> None:
         if result != TDB_SUCCESS:
             raise TidesDBError.from_code(result, "failed to create backup")
 
+    def checkpoint(self, checkpoint_dir: str) -> None:
+        """
+        Create a lightweight, near-instant snapshot of the database using hard links.
+
+        Unlike backup(), checkpoint uses hard links instead of copying SSTable data,
+        making it near-instant and using no extra disk space until compaction removes
+        old SSTables.
+
+        Args:
+            checkpoint_dir: Path to the checkpoint directory (must be non-existent or empty)
+
+        Raises:
+            TidesDBError: If checkpoint fails (e.g., directory not empty, I/O error)
+        """
+        if self._closed:
+            raise TidesDBError("Database is closed")
+
+        result = _lib.tidesdb_checkpoint(self._db, checkpoint_dir.encode("utf-8"))
+        if result != TDB_SUCCESS:
+            raise TidesDBError.from_code(result, "failed to create checkpoint")
+
     def rename_column_family(self, old_name: str, new_name: str) -> None:
         """
         Atomically rename a column family and its underlying directory.

tests/test_tidesdb.py

@@ -529,5 +529,65 @@ def test_compact(self, db, cf):
         time.sleep(0.5)
 
 
+class TestCheckpoint:
+    """Tests for checkpoint operations."""
+
+    def test_checkpoint_creates_snapshot(self, db, cf, temp_db_path):
+        """Test that checkpoint creates a usable snapshot."""
+        with db.begin_txn() as txn:
+            txn.put(cf, b"key1", b"value1")
+            txn.put(cf, b"key2", b"value2")
+            txn.commit()
+
+        checkpoint_dir = temp_db_path + "_checkpoint"
+        try:
+            db.checkpoint(checkpoint_dir)
+            assert os.path.isdir(checkpoint_dir)
+
+            with tidesdb.TidesDB.open(checkpoint_dir) as checkpoint_db:
+                cp_cf = checkpoint_db.get_column_family("test_cf")
+                with checkpoint_db.begin_txn() as txn:
+                    assert txn.get(cp_cf, b"key1") == b"value1"
+                    assert txn.get(cp_cf, b"key2") == b"value2"
+        finally:
+            shutil.rmtree(checkpoint_dir, ignore_errors=True)
+
+    def test_checkpoint_existing_dir_raises(self, db, cf, temp_db_path):
+        """Test that checkpoint to a non-empty directory raises error."""
+        with db.begin_txn() as txn:
+            txn.put(cf, b"key1", b"value1")
+            txn.commit()
+
+        checkpoint_dir = temp_db_path + "_checkpoint"
+        try:
+            db.checkpoint(checkpoint_dir)
+
+            with pytest.raises(tidesdb.TidesDBError):
+                db.checkpoint(checkpoint_dir)
+        finally:
+            shutil.rmtree(checkpoint_dir, ignore_errors=True)
+
+    def test_checkpoint_independence(self, db, cf, temp_db_path):
+        """Test that checkpoint is independent from the live database."""
+        with db.begin_txn() as txn:
+            txn.put(cf, b"key1", b"original")
+            txn.commit()
+
+        checkpoint_dir = temp_db_path + "_checkpoint"
+        try:
+            db.checkpoint(checkpoint_dir)
+
+            with db.begin_txn() as txn:
+                txn.put(cf, b"key1", b"modified")
+                txn.commit()
+
+            with tidesdb.TidesDB.open(checkpoint_dir) as checkpoint_db:
+                cp_cf = checkpoint_db.get_column_family("test_cf")
+                with checkpoint_db.begin_txn() as txn:
+                    assert txn.get(cp_cf, b"key1") == b"original"
+        finally:
+            shutil.rmtree(checkpoint_dir, ignore_errors=True)
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])