localvectordb.backup module

Backup, restore, and recovery system for LocalVectorDB.

This module provides comprehensive backup and recovery capabilities for LocalVectorDB, including full backups, incremental backups, and point-in-time recovery. The system leverages SQLite’s built-in backup API and FAISS’s save/load functionality.

Classes:: BackupManager: Core backup and restore functionality BackupMetadata: Backup metadata structure BackupConfig: Configuration for backup operations

class localvectordb.backup.BackupType(*values)

Bases: Enum

Backup type enumeration.

FULL = 'full'

INCREMENTAL = 'incremental'

class localvectordb.backup.CompressionAlgorithm(*values)

Bases: Enum

Supported compression algorithms.

NONE = 'none'

GZIP = 'gzip'

LZMA = 'lzma'

BZIP = 'bzip2'

class localvectordb.backup.BackupMetadata(backup_id: str, backup_type: BackupType, database_name: str, database_version: str, created_at: datetime, file_paths: Dict[str, str], checksums: Dict[str, str], compression_algorithm: CompressionAlgorithm = CompressionAlgorithm.GZIP, size_bytes: int = 0, parent_backup_id: str | None = None, metadata: Dict[str, Any] | None = None, archive_checksum: str | None = None, manifest_checksum: str | None = None)

Bases: object

Metadata for backup files.

Contains information about the backup including version, timestamp, checksums, and configuration details needed for restoration.

Parameters:

backup_id (str) – Unique identifier for the backup
backup_type (BackupType) – Type of backup (full or incremental)
database_name (str) – Name of the source database
database_version (str) – Version of the source database schema
created_at (datetime) – Timestamp when backup was created
file_paths (Dict[str, str]) – Mapping of component names to file paths in the backup
checksums (Dict[str, str]) – SHA-256 checksums for each component
compression_algorithm (CompressionAlgorithm) – Compression algorithm used
size_bytes (int) – Total size of the backup in bytes
parent_backup_id (str, optional) – ID of parent backup (for incremental backups)
metadata (Dict[str, Any], optional) – Additional metadata
archive_checksum (str, optional) – SHA-256 checksum of the entire archive file for integrity verification
manifest_checksum (str, optional) – SHA-256 checksum of the manifest.json for tampering detection

__init__(backup_id: str, backup_type: BackupType, database_name: str, database_version: str, created_at: datetime, file_paths: Dict[str, str], checksums: Dict[str, str], compression_algorithm: CompressionAlgorithm = CompressionAlgorithm.GZIP, size_bytes: int = 0, parent_backup_id: str | None = None, metadata: Dict[str, Any] | None = None, archive_checksum: str | None = None, manifest_checksum: str | None = None)

to_dict() → Dict[str, Any]: Convert metadata to dictionary for JSON serialization.

classmethod from_dict(data: Dict[str, Any]) → BackupMetadata: Create BackupMetadata from dictionary.

class localvectordb.backup.BackupConfig(backup_location: str | Path = './backups', compression_algorithm: CompressionAlgorithm = CompressionAlgorithm.GZIP, verify_integrity: bool = True, retention_days: int = 30, max_backup_size_gb: float = 0.0, include_faiss_index: bool = True)

Bases: object

Configuration for backup operations.

Parameters:

backup_location (Union[str, Path]) – Directory to store backup files
compression_algorithm (CompressionAlgorithm) – Compression algorithm to use
verify_integrity (bool) – Whether to verify backup integrity after creation
retention_days (int) – Number of days to retain backups
max_backup_size_gb (float) – Maximum backup size in GB (0 = unlimited)
include_faiss_index (bool) – Whether to include FAISS index in backups

__init__(backup_location: str | Path = './backups', compression_algorithm: CompressionAlgorithm = CompressionAlgorithm.GZIP, verify_integrity: bool = True, retention_days: int = 30, max_backup_size_gb: float = 0.0, include_faiss_index: bool = True)

backup_location: Path

Bases: object

Main backup and restore manager for LocalVectorDB.

Provides comprehensive backup and recovery capabilities including full backups, incremental backups, and point-in-time recovery using SQLite’s backup API and FAISS index management.

Parameters:

database_path (Union[str, Path]) – Path to the LocalVectorDB database file
faiss_index_path (Union[str, Path], optional) – Path to the FAISS index file. If None, inferred from database_path.
config (BackupConfig, optional) – Backup configuration. If None, uses default configuration.
db (LocalVectorDB, optional) –
The live database these paths belong to. Pass this to back up a database that is open and may be written to concurrently: the backup then holds the database’s write lock and flushes the FAISS index to disk for the duration of the snapshot, so the SQLite copy and the index copy are mutually consistent.

Without it (the path-only form) the two stores are copied without coordination. A backup taken while a write is in flight can capture SQLite rows whose vectors are not yet in the copied index – i.e. dangling rows. So the path-only form is safe only for a database that is closed, or that is otherwise known to be quiescent.

Examples

Create a full backup of a closed database:

manager = BackupManager("/path/to/mydb.sqlite")
backup_id = manager.create_backup(BackupType.FULL)
print(f"Backup created: {backup_id}")

Create a consistent backup of a live, open database:

manager = BackupManager(db.db_path, db=db)
backup_id = manager.create_backup(BackupType.FULL)

Restore from backup:

manager.restore_backup(backup_id, "/path/to/restore/location")

List available backups:

backups = manager.list_backups()
for backup in backups:
    print(f"{backup.backup_id}: {backup.created_at}")

__init__(database_path: str | Path, faiss_index_path: str | Path | None = None, config: BackupConfig | None = None, db: Any | None = None)

create_backup(backup_type: BackupType = BackupType.FULL, parent_backup_id: str | None = None, backup_id: str | None = None) → str

Create a new backup of the database.

Parameters:

backup_type (BackupType) – Type of backup to create
parent_backup_id (str, optional) – ID of parent backup (required for incremental backups)
backup_id (str, optional) – Custom backup ID. If None, generates a UUID.

Returns:

Unique backup ID

Return type:

str

Raises:

FileNotFoundError – If database file doesn’t exist
ValueError – If incremental backup requested without parent ID

list_backups(backup_type: BackupType | None = None) → List[BackupMetadata]

List available backups.

Parameters:: backup_type (BackupType, optional) – Filter by backup type. If None, returns all backups.
Returns:: List of backup metadata objects
Return type:: List[BackupMetadata]

restore_backup(backup_id: str, restore_location: str | Path | None = None, overwrite_existing: bool = False) → Path

Restore database from backup.

Parameters:

backup_id (str) – ID of backup to restore
restore_location (Union[str, Path], optional) – Directory to restore to. If None, restores to original location.
overwrite_existing (bool) – Whether to overwrite existing files

Returns:

Path to restored database directory

Return type:

Path

Raises:

FileNotFoundError – If backup file not found
ValueError – If restore would overwrite existing files without permission

MAX_PATH_LENGTH = 4096

delete_backup(backup_id: str) → bool

Delete a backup.

Parameters:: backup_id (str) – ID of backup to delete
Returns:: True if backup was deleted successfully
Return type:: bool

cleanup_old_backups(retention_days: int | None = None) → int

Clean up old backups based on retention policy.

Parameters:: retention_days (int, optional) – Number of days to retain backups. If None, uses config value.
Returns:: Number of backups deleted
Return type:: int

verify_backup_streaming(backup_id: str, verify_archive_members: bool = True) → bool

Verify backup integrity using streaming without full extraction.

Parameters:

backup_id (str) – ID of backup to verify
verify_archive_members (bool, default True) – Whether to verify individual archive member checksums. If False, only verifies archive and manifest checksums.

Returns:

True if backup is valid

Return type:

bool

Notes

This method provides efficient verification by streaming archive contents rather than extracting to disk. Useful for large backups or when disk space is limited.

verify_backup(backup_id: str) → bool

Verify backup integrity without restoring.

Parameters:: backup_id (str) – ID of backup to verify
Returns:: True if backup is valid
Return type:: bool

get_backup_info(backup_id: str) → BackupMetadata | None

Get detailed information about a backup.

Parameters:: backup_id (str) – ID of backup to get info for
Returns:: Backup metadata if found
Return type:: BackupMetadata or None

class localvectordb.backup.IncrementalBackupManager(backup_manager: BackupManager)

Bases: object

Specialized manager for incremental backups using WAL tracking.

Implements incremental backup functionality by tracking changes in SQLite’s Write-Ahead Log (WAL) and maintaining FAISS index deltas.

Parameters:: backup_manager (BackupManager) – Parent backup manager instance

__init__(backup_manager: BackupManager)

create_incremental_backup(parent_backup_id: str, backup_id: str | None = None) → str

Create an incremental backup based on changes since parent backup.

Parameters:

parent_backup_id (str) – ID of the parent (full or incremental) backup
backup_id (str, optional) – Custom backup ID. If None, generates a UUID.

Returns:

Unique backup ID for the incremental backup

Return type:

str

Raises:

ValueError – If parent backup not found or WAL mode not enabled
FileNotFoundError – If database files don’t exist

restore_incremental_backup_chain(target_backup_id: str, restore_location: str | Path) → Path

Restore database by applying a chain of incremental backups.

Parameters:

target_backup_id (str) – ID of the target backup (can be full or incremental)
restore_location (Union[str, Path]) – Directory to restore to

Returns:

Path to restored database directory

Return type:

Path

class localvectordb.backup.PointInTimeRecoveryManager(backup_manager: BackupManager, incremental_manager: IncrementalBackupManager)

Bases: object

Point-in-time recovery (PITR) manager for LocalVectorDB.

Provides the ability to restore a database to any point in time within the backup retention window by combining full and incremental backups.

Parameters:

backup_manager (BackupManager) – Parent backup manager instance
incremental_manager (IncrementalBackupManager) – Incremental backup manager instance

__init__(backup_manager: BackupManager, incremental_manager: IncrementalBackupManager)

get_recovery_timeline() → List[Dict[str, Any]]

Get the available recovery timeline showing all recovery points.

Returns:: List of recovery points with timestamps and backup information
Return type:: List[Dict[str, Any]]

find_recovery_point(target_timestamp: datetime, tolerance_minutes: int = 60) → Dict[str, Any] | None

Find the best recovery point for a target timestamp.

Parameters:

target_timestamp (datetime) – Target timestamp for recovery
tolerance_minutes (int) – Maximum tolerance in minutes to find a recovery point

Returns:

Recovery point information if found

Return type:

Dict[str, Any] or None

restore_to_point_in_time(target_timestamp: datetime, restore_location: str | Path, tolerance_minutes: int = 60, dry_run: bool = False) → Dict[str, Any]

Restore database to a specific point in time.

Parameters:

target_timestamp (datetime) – Target timestamp for recovery
restore_location (Union[str, Path]) – Directory to restore to
tolerance_minutes (int) – Maximum tolerance in minutes to find a recovery point
dry_run (bool) – If True, only validate the recovery without actually restoring

Returns:

Recovery operation results with status and details

Return type:

Dict[str, Any]

validate_recovery_timeline() → Dict[str, Any]

Validate the recovery timeline for consistency and completeness.

Returns:: Validation results with any issues found
Return type:: Dict[str, Any]

cleanup_recovery_timeline(max_age_days: int | None = None, keep_full_backups: int = 3, dry_run: bool = False) → Dict[str, Any]

Clean up old backups while maintaining recovery timeline integrity.

Parameters:

max_age_days (int, optional) – Maximum age of backups to keep. If None, uses config retention_days.
keep_full_backups (int) – Minimum number of full backups to keep regardless of age
dry_run (bool) – If True, only simulate cleanup without actually deleting

Returns:

Cleanup operation results

Return type:

Dict[str, Any]

get_recovery_recommendations(target_timestamp: datetime) → Dict[str, Any]

Get recommendations for recovering to a specific point in time.

Parameters:: target_timestamp (datetime) – Target timestamp for recovery
Returns:: Recovery recommendations and analysis
Return type:: Dict[str, Any]