Skip to content

type_bridge.migration.generator

generator

Migration generator for auto-generating migrations from model changes.

The generator is a thin adapter over the canonical Rust authoring core (#166): it introspects the live base schema, collects the target schema from the registered models, and delegates diffing, operation mapping, .py/sidecar rendering, and snapshot generation to :func:type_bridge.migration.author.author_migration. Live and offline authoring therefore share exactly one SchemaDiff -> operations mapping.

MigrationGenerator 

MigrationGenerator(db, migrations_dir)

Generates migration files from model changes.

Compares current models against the introspected database schema and delegates artifact authoring to the shared Rust core.

Example

generator = MigrationGenerator(db, Path("migrations"))

Generate migration from models

path = generator.generate([Person, Company, Employment], name="initial")

Creates: migrations/0001_initial.py

Generate empty migration for manual editing

path = generator.generate([], name="custom_changes", empty=True)

Initialize generator.

Parameters:

Name Type Description Default
db Database

Database connection

 required
migrations_dir Path

Directory to write migration files

 required
Source code in type_bridge/migration/generator.py 
49
50
51
52
53
54
55
56
57
58
def __init__(self, db: Database, migrations_dir: Path):
    """Initialize generator.

    Args:
        db: Database connection
        migrations_dir: Directory to write migration files
    """
    self.db = db
    self.migrations_dir = migrations_dir
    self.loader = MigrationLoader(migrations_dir)

generate 

generate(models, name='auto', empty=False)

Generate a migration file.

Parameters:

Name Type Description Default
models list[type[Entity | Relation]]

Model classes to check for changes

 required
name str

Migration name suffix (e.g., "initial", "add_company")

 'auto'
empty bool

Create empty migration for manual editing

 False

Returns:

Type Description
Path | None

Path to created file, or None if no changes detected
Source code in type_bridge/migration/generator.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def generate(
    self,
    models: list[type[Entity | Relation]],
    name: str = "auto",
    empty: bool = False,
) -> Path | None:
    """Generate a migration file.

    Args:
        models: Model classes to check for changes
        name: Migration name suffix (e.g., "initial", "add_company")
        empty: Create empty migration for manual editing

    Returns:
        Path to created file, or None if no changes detected
    """
    existing = self.loader.discover()
    next_num = self.loader.get_next_number()

    dependencies: list[tuple[str, str]] = []
    if existing:
        last = existing[-1]
        dependencies.append((last.migration.app_label, last.migration.name))

    if empty:
        return self._generate_empty(name, next_num, dependencies)

    if not models:
        logger.info("No changes detected")
        return None

    # Collect the target schema from models.
    schema_mgr = SchemaManager(self.db)
    schema_mgr.register(*models)
    new_info = schema_mgr.collect_schema_info()

    # Introspect the full application schema so removed target types are
    # still visible to the diff. TypeBridge's own migration state schema is
    # filtered out because it is storage infrastructure, not app schema.
    introspector = SchemaIntrospector(self.db)
    db_schema = without_migration_state_schema(introspector.introspect())

    authored = author_migration(
        db_schema.to_rust_schema_info(),
        new_info.to_rust_schema_info(),
        app_label=self.migrations_dir.name,
        name=f"{next_num:04d}_{name}",
        dependencies=dependencies,
        snapshot_version=f"v{next_num:04d}",
        previous_snapshot_version=f"v{next_num - 1:04d}" if next_num > 1 else None,
    )
    if authored is None:
        logger.info("No changes detected")
        return None

    self.migrations_dir.mkdir(parents=True, exist_ok=True)
    filepath = authored.write_to(self.migrations_dir)
    logger.info(f"Created migration: {filepath}")
    return filepath