Skip to content

conflict_strategy

Strategies for handling migration conflicts during validation.

IgnoreConflict

Bases: MigrationConflictStrategy

Silently ignores modified or removed migrations and continues execution.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
74
75
76
77
78
79
80
81
class IgnoreConflict(MigrationConflictStrategy):
    """Silently ignores modified or removed migrations and continues execution."""

    def on_modified(self, migrations: list[Migration]) -> None:
        """Do nothing."""

    def on_removed(self, migrations: list[Migration]) -> None:
        """Do nothing."""

on_modified(migrations)

Do nothing.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
77
78
def on_modified(self, migrations: list[Migration]) -> None:
    """Do nothing."""

on_removed(migrations)

Do nothing.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
80
81
def on_removed(self, migrations: list[Migration]) -> None:
    """Do nothing."""

MigrationConflictStrategy

Bases: ABC

Abstract base class defining how to react to modified or removed migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class MigrationConflictStrategy(ABC):
    """Abstract base class defining how to react to modified or removed migrations."""

    @abstractmethod
    def on_modified(self, migrations: list[Migration]) -> None:
        """Handle migrations that have been modified since the last execution.

        Args:
            migrations: Modified migrations detected by the runner.

        """

    @abstractmethod
    def on_removed(self, migrations: list[Migration]) -> None:
        """Handle migrations that have been removed since the last execution.

        Args:
            migrations: Removed migrations detected by the runner.

        """

on_modified(migrations) abstractmethod

Handle migrations that have been modified since the last execution.

Parameters:

Name Type Description Default
migrations list[Migration]

Modified migrations detected by the runner.

 required
Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
14
15
16
17
18
19
20
21
@abstractmethod
def on_modified(self, migrations: list[Migration]) -> None:
    """Handle migrations that have been modified since the last execution.

    Args:
        migrations: Modified migrations detected by the runner.

    """

on_removed(migrations) abstractmethod

Handle migrations that have been removed since the last execution.

Parameters:

Name Type Description Default
migrations list[Migration]

Removed migrations detected by the runner.

 required
Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
23
24
25
26
27
28
29
30
@abstractmethod
def on_removed(self, migrations: list[Migration]) -> None:
    """Handle migrations that have been removed since the last execution.

    Args:
        migrations: Removed migrations detected by the runner.

    """

RaiseOnConflict

Bases: MigrationConflictStrategy

Raises ValueError when modified or removed migrations are detected.

This is the default behaviour, preserving strict integrity guarantees.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class RaiseOnConflict(MigrationConflictStrategy):
    """Raises ValueError when modified or removed migrations are detected.

    This is the default behaviour, preserving strict integrity guarantees.
    """

    def on_modified(self, migrations: list[Migration]) -> None:
        """Raise ValueError listing all modified migrations."""
        if migrations:
            migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
            raise ValueError(f"The following migrations have been modified since the last execution:\n{migration_str}")

    def on_removed(self, migrations: list[Migration]) -> None:
        """Raise ValueError listing all removed migrations."""
        if migrations:
            migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
            raise ValueError(f"The following migrations have been removed since the last execution:\n{migration_str}")

on_modified(migrations)

Raise ValueError listing all modified migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
39
40
41
42
43
def on_modified(self, migrations: list[Migration]) -> None:
    """Raise ValueError listing all modified migrations."""
    if migrations:
        migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
        raise ValueError(f"The following migrations have been modified since the last execution:\n{migration_str}")

on_removed(migrations)

Raise ValueError listing all removed migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
45
46
47
48
49
def on_removed(self, migrations: list[Migration]) -> None:
    """Raise ValueError listing all removed migrations."""
    if migrations:
        migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
        raise ValueError(f"The following migrations have been removed since the last execution:\n{migration_str}")

WarnOnConflict

Bases: MigrationConflictStrategy

Emits a :mod:warnings warning and continues when conflicts are detected.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class WarnOnConflict(MigrationConflictStrategy):
    """Emits a :mod:`warnings` warning and continues when conflicts are detected."""

    def on_modified(self, migrations: list[Migration]) -> None:
        """Emit a warning listing all modified migrations."""
        if migrations:
            migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
            warnings.warn(
                f"The following migrations have been modified since the last execution:\n{migration_str}",
                stacklevel=3,
            )

    def on_removed(self, migrations: list[Migration]) -> None:
        """Emit a warning listing all removed migrations."""
        if migrations:
            migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
            warnings.warn(
                f"The following migrations have been removed since the last execution:\n{migration_str}",
                stacklevel=3,
            )

on_modified(migrations)

Emit a warning listing all modified migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
55
56
57
58
59
60
61
62
def on_modified(self, migrations: list[Migration]) -> None:
    """Emit a warning listing all modified migrations."""
    if migrations:
        migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
        warnings.warn(
            f"The following migrations have been modified since the last execution:\n{migration_str}",
            stacklevel=3,
        )

on_removed(migrations)

Emit a warning listing all removed migrations.

Source code in src/cloe_delta_table_manager/migrations/conflict_strategy.py 
64
65
66
67
68
69
70
71
def on_removed(self, migrations: list[Migration]) -> None:
    """Emit a warning listing all removed migrations."""
    if migrations:
        migration_str = _MIGRATION_SEPARATOR.join(str(m) for m in migrations)
        warnings.warn(
            f"The following migrations have been removed since the last execution:\n{migration_str}",
            stacklevel=3,
        )