migration

Module for defining migration and migration state classes.

FileMigration

Bases: Migration

Represents a user-defined migration script with additional metadata.

Source code in src/cloe_delta_table_manager/migrations/migration.py

class FileMigration(Migration):
    """Represents a user-defined migration script with additional metadata."""

    depends_on_paths: list[Path] = Field(default_factory=list)
    source_path: Path

    @field_validator("depends_on_paths", mode="before")
    @classmethod
    def validate_depends_on_path(cls, v):
        """Validate that depends_on_paths contains only relative paths."""
        paths = [Path(item) if isinstance(item, str) else item for item in v]
        for path in paths:
            if path.is_absolute():
                raise ValueError(f"Absolute paths are not allowed in depends_on_paths: {path}")
        return paths

    @model_validator(mode="after")
    def resolve_depends_on_paths(self):
        """Resolve depends_on_paths relative to source_path directory."""
        if self.depends_on_paths and self.source_path:
            source_dir = Path(self.source_path).parent
            self.depends_on_paths = [source_dir / path for path in self.depends_on_paths]
        return self

resolve_depends_on_paths()

Resolve depends_on_paths relative to source_path directory.

Source code in src/cloe_delta_table_manager/migrations/migration.py

@model_validator(mode="after")
def resolve_depends_on_paths(self):
    """Resolve depends_on_paths relative to source_path directory."""
    if self.depends_on_paths and self.source_path:
        source_dir = Path(self.source_path).parent
        self.depends_on_paths = [source_dir / path for path in self.depends_on_paths]
    return self

validate_depends_on_path(v) classmethod

Validate that depends_on_paths contains only relative paths.

Source code in src/cloe_delta_table_manager/migrations/migration.py

@field_validator("depends_on_paths", mode="before")
@classmethod
def validate_depends_on_path(cls, v):
    """Validate that depends_on_paths contains only relative paths."""
    paths = [Path(item) if isinstance(item, str) else item for item in v]
    for path in paths:
        if path.is_absolute():
            raise ValueError(f"Absolute paths are not allowed in depends_on_paths: {path}")
    return paths

Migration

Bases: BaseModel

Represents a database migration with operations and dependencies.

Source code in src/cloe_delta_table_manager/migrations/migration.py

class Migration(BaseModel):
    """Represents a database migration with operations and dependencies."""

    operations: list[Operation]
    status: MigrationStatus = MigrationStatus.NEW
    depends_on: list[Self] = Field(default_factory=list)

    @computed_field
    @property
    def operation_hash(self) -> int:
        """Compute a deterministic migration id as a SHA-256 hash of all operations.

        Uses hashlib instead of Python's built-in hash() to guarantee identical
        values across processes regardless of PYTHONHASHSEED.
        """
        if not self.operations:
            return 0
        content = "|".join(f"{op.index}:{op.operation_type.value}:{';'.join(op.commands)}" for op in self.operations)
        digest = hashlib.sha256(content.encode()).digest()
        return int.from_bytes(digest[:8], "big")

    @classmethod
    def from_diff(
        cls,
        diff: Diff,
        on_unsupported: OnUnsupported = "fail",
        platform_options: AzureDatabricksOptions | None = None,
    ) -> Migration:
        """Create a Migration object from a typed Diff object."""
        from cloe_delta_table_manager.diff import (  # noqa: PLC0415
            DatabricksStatementGenerator,
        )

        operation = Operation(
            commands=DatabricksStatementGenerator(
                on_unsupported=on_unsupported,
                options=platform_options,
            ).generate(diff),
            operation_type=_operation_type_for(diff),
            index=0,
        )
        return cls(operations=[operation], depends_on=[])

    def __str__(self) -> str:
        """Migration string representation showing id and status."""
        return f"Migration(id={self.operation_hash}, status={self.status})"

    def __repr__(self) -> str:
        """Detailed string representation showing id, status, number of operations, and dependencies."""
        return f"Migration(id={self.operation_hash}, status={self.status}, operations={len(self.operations)}, depends_on={len(self.depends_on)})"

operation_hash property

Compute a deterministic migration id as a SHA-256 hash of all operations.

Uses hashlib instead of Python's built-in hash() to guarantee identical values across processes regardless of PYTHONHASHSEED.

__repr__()

Detailed string representation showing id, status, number of operations, and dependencies.

Source code in src/cloe_delta_table_manager/migrations/migration.py

def __repr__(self) -> str:
    """Detailed string representation showing id, status, number of operations, and dependencies."""
    return f"Migration(id={self.operation_hash}, status={self.status}, operations={len(self.operations)}, depends_on={len(self.depends_on)})"

__str__()

Migration string representation showing id and status.

Source code in src/cloe_delta_table_manager/migrations/migration.py

def __str__(self) -> str:
    """Migration string representation showing id and status."""
    return f"Migration(id={self.operation_hash}, status={self.status})"

from_diff(diff, on_unsupported='fail', platform_options=None) classmethod

Create a Migration object from a typed Diff object.

Source code in src/cloe_delta_table_manager/migrations/migration.py

@classmethod
def from_diff(
    cls,
    diff: Diff,
    on_unsupported: OnUnsupported = "fail",
    platform_options: AzureDatabricksOptions | None = None,
) -> Migration:
    """Create a Migration object from a typed Diff object."""
    from cloe_delta_table_manager.diff import (  # noqa: PLC0415
        DatabricksStatementGenerator,
    )

    operation = Operation(
        commands=DatabricksStatementGenerator(
            on_unsupported=on_unsupported,
            options=platform_options,
        ).generate(diff),
        operation_type=_operation_type_for(diff),
        index=0,
    )
    return cls(operations=[operation], depends_on=[])

MigrationStatus

Bases: Enum

Enum representing the status of a migration.

Source code in src/cloe_delta_table_manager/migrations/migration.py

class MigrationStatus(Enum):
    """Enum representing the status of a migration."""

    UNDEFINED = "undefined"
    NEW = "new"
    PENDING = "pending"
    APPLIED = "applied"
    FAILED = "failed"