localvectordb.migration module

Metadata schema migration system for LocalVectorDB.

This module provides comprehensive metadata schema migration capabilities for LocalVectorDB. It focuses on evolving the metadata schema using the existing DatabaseSchema functionality rather than raw SQL operations.

The migration system handles: - Adding, removing, and modifying metadata fields - Column remapping and data migration - Schema validation and compatibility checking - Version tracking and rollback functionality

Classes:

Migration: Base class for metadata schema migrations MigrationEngine: Core migration management engine MigrationScript: Individual migration script representation

localvectordb.migration.serialize_metadata_field(field: MetadataField) → Dict[str, Any]

Serialize a MetadataField object to a JSON-compatible dictionary.

localvectordb.migration.deserialize_metadata_field(data: Dict[str, Any]) → MetadataField

Deserialize a dictionary back to a MetadataField object.

localvectordb.migration.serialize_schema_changes(changes: Dict[str, Any]) → str

Serialize schema changes containing MetadataField objects to JSON string.

localvectordb.migration.deserialize_schema_changes(json_str: str) → Dict[str, Any]

Deserialize JSON string back to schema changes with MetadataField objects.

class localvectordb.migration.Migration(database_path: str | Path)

Bases: ABC

Abstract base class for metadata schema migrations.

All migration classes must inherit from this base class and implement the get_schema_changes() and get_rollback_changes() methods for forward and backward migrations.

This class focuses on metadata schema evolution rather than raw SQL operations, leveraging the existing DatabaseSchema functionality.

Variables:

• version (str) – Target version for this migration

• description (str) – Human-readable description of the migration

• dependencies (List[str]) – List of migration versions that must be applied before this one

version: str = '0.0.0'

description: str = 'Base migration'

dependencies: List[str] = []

__init__(database_path: str | Path)

abstractmethod get_schema_changes() → Dict[str, Any]

Get the schema changes to apply in the forward migration.

Returns:

Schema changes specification with the following structure:

{
    'new_schema': Dict[str, MetadataField],  # Complete new schema
    'column_mapping': Dict[str, str],        # Rename mappings: old -> new
    'drop_columns': bool                     # Whether to drop unused columns
}

Return type:

Dict[str, Any]

Examples

Add new fields:

return {
    'new_schema': {
        'user_id': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'priority': MetadataField(type=MetadataFieldType.INTEGER, default_value=0)
    }
}

Rename and modify fields:

return {
    'new_schema': {
        'author_name': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'created_date': MetadataField(type=MetadataFieldType.DATE, indexed=True)
    },
    'column_mapping': {
        'author': 'author_name',
        'timestamp': 'created_date'
    }
}

abstractmethod get_rollback_changes() → Dict[str, Any]

Get the schema changes to apply for rollback.

Returns:

Schema changes specification for rolling back this migration

Return type:

Dict[str, Any]

validate_prerequisites(current_schema: Dict[str, MetadataField]) → bool

Validate that prerequisites for this migration are met.

Parameters:

current_schema (Dict[str, MetadataField]) – Current metadata schema

Returns:

True if prerequisites are met

Return type:

bool

class localvectordb.migration.MigrationScript(file_path: Path, migration_class: Type[Migration])

Bases: object

Represents a single migration script file.

Parameters:

• file_path (Path) – Path to the migration script file

• migration_class (Type[Migration]) – Migration class loaded from the file

__init__(file_path: Path, migration_class: Type[Migration])

create_instance(database_path: str | Path) → Migration

Create an instance of the migration class.

class localvectordb.migration.MigrationEngine(database_path: str | Path, migrations_directory: str | Path = './migrations', backup_manager: BackupManager | None = None, auto_backup: bool = True)

Bases: object

Core database migration management engine.

Handles discovery, validation, execution, and rollback of database migrations. Integrates with the backup system for safety and the versioning system for tracking applied migrations.

Parameters:

• database_path (Union[str, Path]) – Path to the database file

• migrations_directory (Union[str, Path], optional) – Directory containing migration scripts. Defaults to “./migrations”

• backup_manager (BackupManager, optional) – Backup manager for creating safety backups before migrations

• auto_backup (bool) – Whether to automatically create backups before migrations

__init__(database_path: str | Path, migrations_directory: str | Path = './migrations', backup_manager: BackupManager | None = None, auto_backup: bool = True)

discover_migrations() → Dict[str, MigrationScript]

Discover and load all migration scripts from the migrations directory.

Returns:

Dictionary mapping version strings to MigrationScript objects

Return type:

Dict[str, MigrationScript]

get_migration_order() → List[str]

Get the correct order for applying migrations based on dependencies.

Returns:

List of version strings in the order they should be applied

Return type:

List[str]

Raises:

ValueError – If circular dependencies are detected

get_applied_migrations() → List[Dict[str, Any]]

Get list of migrations that have been applied to the database.

Returns:

List of applied migration records

Return type:

List[Dict[str, Any]]

get_pending_migrations(target_version: str | None = None) → List[str]

Get list of migrations that need to be applied.

Parameters:

target_version (str, optional) – Target version to migrate to. If None, migrates to latest.

Returns:

List of migration versions that need to be applied

Return type:

List[str]

migrate(target_version: str | None = None, dry_run: bool = False, create_backup: bool | None = None) → Dict[str, Any]

Apply pending migrations up to the target version.

Parameters:

• target_version (str, optional) – Target version to migrate to. If None, migrates to latest.

• dry_run (bool) – If True, validate migrations without applying them

• create_backup (bool, optional) – Whether to create a backup before migration. If None, uses auto_backup setting.

Returns:

Migration results with status and details

Return type:

Dict[str, Any]

rollback(target_version: str, dry_run: bool = False, create_backup: bool | None = None) → Dict[str, Any]

Rollback migrations to a target version.

Parameters:

• target_version (str) – Target version to rollback to

• dry_run (bool) – If True, validate rollback without applying it

• create_backup (bool, optional) – Whether to create a backup before rollback. If None, uses auto_backup setting.

Returns:

Rollback results with status and details

Return type:

Dict[str, Any]

get_migration_status() → Dict[str, Any]

Get comprehensive status of database migrations.

Returns:

Migration status information

Return type:

Dict[str, Any]

create_migration_template(version: str, description: str, template_type: str = 'basic') → Path

Create a new migration template file.

Parameters:

• version (str) – Version for the new migration

• description (str) – Description of the migration

• template_type (str) – Type of template to create (“basic”, “schema”, “data”)

Returns:

Path to the created migration file

Return type:

Path