Migration Hooks

Here's the reference information for the included migration hooks classes, with all their parameters, attributes and methods.

You can import these classes directly from pyrmute:

from pyrmute import MigrationHook, MetricsHook

pyrmute.MigrationHook

Base class for migration hooks.

Hooks are read-only observers that allow you to inject custom behavior before, after, or on error during migrations. Useful for logging, monitoring, metrics, validation, or auditing.

Important

Hook methods receive read-only views of data and should NOT modify it. Data transformation should only happen in migration functions.

Example

class LoggingHook(MigrationHook):
    def before_migrate(
        self,
        name: str,
        from_version: ModelVersion,
        to_version: ModelVersion,
        data: Mapping[str, Any],
    ) -> None:
        logger.info(f"Migrating {name} from {from_version} to {to_version}")

    def after_migrate(
        self,
        name: str,
        from_version: ModelVersion,
        to_version: ModelVersion,
        original_data: Mapping[str, Any],
        migrated_data: Mapping[str, Any],
    ) -> None:
        logger.info(f"Successfully migrated {name}")

    def on_error(
        self,
        name: str,
        from_version: ModelVersion,
        to_version: ModelVersion,
        data: Mapping[str, Any],
        error: Exception,
    ) -> None:
        logger.error(f"Migration failed: {error}")

before_migrate

before_migrate(name, from_version, to_version, data)

Called before migration starts.

PARAMETER	DESCRIPTION
name	Model name being migrated. TYPE: str
from_version	Source version. TYPE: ModelVersion
to_version	Target version. TYPE: ModelVersion
data	Original data to be migrated (read-only view). TYPE: Mapping[str, Any]

Warning

Do not modify the data parameter. Hooks are for observation only.

Source code in src/pyrmute/migration_hooks.py

def before_migrate(
    self,
    name: str,
    from_version: ModelVersion,
    to_version: ModelVersion,
    data: Mapping[str, Any],
) -> None:
    """Called before migration starts.

    Args:
        name: Model name being migrated.
        from_version: Source version.
        to_version: Target version.
        data: Original data to be migrated (read-only view).

    Warning:
        Do not modify the data parameter. Hooks are for observation only.
    """

after_migrate

after_migrate(
    name,
    from_version,
    to_version,
    original_data,
    migrated_data,
)

Called after successful migration.

PARAMETER	DESCRIPTION
name	Model name that was migrated. TYPE: str
from_version	Source version. TYPE: ModelVersion
to_version	Target version. TYPE: ModelVersion
original_data	Original input data (read-only view). TYPE: Mapping[str, Any]
migrated_data	Result of migration (read-only view). TYPE: Mapping[str, Any]

Warning

Do not modify the data parameters. Hooks are for observation only.

Source code in src/pyrmute/migration_hooks.py

def after_migrate(
    self,
    name: str,
    from_version: ModelVersion,
    to_version: ModelVersion,
    original_data: Mapping[str, Any],
    migrated_data: Mapping[str, Any],
) -> None:
    """Called after successful migration.

    Args:
        name: Model name that was migrated.
        from_version: Source version.
        to_version: Target version.
        original_data: Original input data (read-only view).
        migrated_data: Result of migration (read-only view).

    Warning:
        Do not modify the data parameters. Hooks are for observation only.
    """

on_error

on_error(name, from_version, to_version, data, error)

Called when migration fails.

PARAMETER	DESCRIPTION
name	Model name that failed to migrate. TYPE: str
from_version	Source version. TYPE: ModelVersion
to_version	Target version. TYPE: ModelVersion
data	Original data that failed to migrate (read-only view). TYPE: Mapping[str, Any]
error	Exception that occurred. TYPE: Exception

Warning

Do not modify the data parameter. Hooks are for observation only.

Source code in src/pyrmute/migration_hooks.py

def on_error(
    self,
    name: str,
    from_version: ModelVersion,
    to_version: ModelVersion,
    data: Mapping[str, Any],
    error: Exception,
) -> None:
    """Called when migration fails.

    Args:
        name: Model name that failed to migrate.
        from_version: Source version.
        to_version: Target version.
        data: Original data that failed to migrate (read-only view).
        error: Exception that occurred.

    Warning:
        Do not modify the data parameter. Hooks are for observation only.
    """

pyrmute.MetricsHook

MetricsHook()

Bases: MigrationHook

Hook for collecting migration metrics.

Tracks migration counts, timing, and errors for monitoring purposes.

Example

metrics = MetricsHook()
manager.add_hook(metrics)

# After migrations
print(f"Total migrations: {metrics.total_count}")
print(f"Failed migrations: {metrics.error_count}")
print(f"Success rate: {metrics.success_rate:.1%}")

Initialize metrics collection.

Source code in src/pyrmute/migration_hooks.py

def __init__(self: Self) -> None:
    """Initialize metrics collection."""
    self.total_count = 0
    self.error_count = 0
    self.migrations_by_model: dict[str, int] = {}
    self.errors_by_model: dict[str, int] = {}

total_count instance-attribute

total_count = 0

error_count instance-attribute

error_count = 0

migrations_by_model instance-attribute

migrations_by_model = {}

errors_by_model instance-attribute

errors_by_model = {}

success_rate property

success_rate

Calculate overall success rate.

RETURNS	DESCRIPTION
float	Success rate as a float between 0 and 1.

after_migrate

after_migrate(
    name,
    from_version,
    to_version,
    original_data,
    migrated_data,
)

Called after successful migration.

PARAMETER	DESCRIPTION
name	Model name that was migrated. TYPE: str
from_version	Source version. TYPE: ModelVersion
to_version	Target version. TYPE: ModelVersion
original_data	Original input data (read-only view). TYPE: Mapping[str, Any]
migrated_data	Result of migration (read-only view). TYPE: Mapping[str, Any]

Warning

Do not modify the data parameters. Hooks are for observation only.

Source code in src/pyrmute/migration_hooks.py

def after_migrate(
    self,
    name: str,
    from_version: ModelVersion,
    to_version: ModelVersion,
    original_data: Mapping[str, Any],
    migrated_data: Mapping[str, Any],
) -> None:
    """Called after successful migration.

    Args:
        name: Model name that was migrated.
        from_version: Source version.
        to_version: Target version.
        original_data: Original input data (read-only view).
        migrated_data: Result of migration (read-only view).

    Warning:
        Do not modify the data parameters. Hooks are for observation only.
    """

before_migrate

before_migrate(name, from_version, to_version, data)

Track migration attempt.

Source code in src/pyrmute/migration_hooks.py

def before_migrate(
    self,
    name: str,
    from_version: ModelVersion,
    to_version: ModelVersion,
    data: Mapping[str, Any],
) -> None:
    """Track migration attempt."""
    self.total_count += 1
    self.migrations_by_model[name] = self.migrations_by_model.get(name, 0) + 1

on_error

on_error(name, from_version, to_version, data, error)

Track migration error.

Source code in src/pyrmute/migration_hooks.py

def on_error(
    self,
    name: str,
    from_version: ModelVersion,
    to_version: ModelVersion,
    data: Mapping[str, Any],
    error: Exception,
) -> None:
    """Track migration error."""
    self.error_count += 1
    self.errors_by_model[name] = self.errors_by_model.get(name, 0) + 1