type_bridge.migration.snapshots

snapshots

Migration snapshot bindings and time-state package management.

SnapshotError

Bases: ValueError

Raised when an existing migration snapshot is missing or stale.

generate_snapshot

generate_snapshot(migrations_dir, version, migration_name, schema_text)

Generate a schema snapshot package in migrations/snapshots/vNNNN/.

Parameters:

Name	Type	Description	Default
migrations_dir	Path	Directory containing migrations (e.g. Path("migrations"))	required
version	str	Snapshot version string, e.g. "v0001"	required
migration_name	str	The suffix name of the migration, e.g. "0001_initial"	required
schema_text	str	The canonical TypeQL schema text	required

Returns:

Type	Description
Path	Path to the created snapshot directory

Source code in type_bridge/migration/snapshots.py

def generate_snapshot(
    migrations_dir: Path,
    version: str,
    migration_name: str,
    schema_text: str,
) -> Path:
    """Generate a schema snapshot package in migrations/snapshots/vNNNN/.

    Args:
        migrations_dir: Directory containing migrations (e.g. Path("migrations"))
        version: Snapshot version string, e.g. "v0001"
        migration_name: The suffix name of the migration, e.g. "0001_initial"
        schema_text: The canonical TypeQL schema text

    Returns:
        Path to the created snapshot directory
    """
    # 1. Establish directory layout
    snapshots_dir = migrations_dir / "snapshots"
    snapshots_dir.mkdir(parents=True, exist_ok=True)

    # Write empty __init__.py in snapshots/ if not exists
    init_file = snapshots_dir / "__init__.py"
    if not init_file.exists():
        init_file.write_text("# TypeBridge migration snapshots package\n")

    schema_hash = _schema_hash(schema_text)
    snapshot_dir = snapshots_dir / version
    # If the snapshot already exists, we do NOT overwrite it (append-only contract)
    if snapshot_dir.exists():
        _validate_existing_snapshot(snapshot_dir, schema_hash)
        return snapshot_dir

    snapshot_dir.mkdir(parents=True, exist_ok=False)

    try:
        # 2. Render Python classes
        generate_models(
            schema=schema_text,
            output_dir=snapshot_dir,
            copy_schema=True,
            schema_path="schema.tql",
        )

        _rewrite_snapshot_init(snapshot_dir)

        # 3. Compute canonical schema hash
        file_hashes = _file_hashes(snapshot_dir)

        # 5. Core version info
        try:
            import type_bridge_core

            core_version = getattr(type_bridge_core, "__version__", None) or getattr(
                type_bridge_core, "VERSION", "unknown"
            )
        except ImportError:
            core_version = "unknown"

        # 6. Render snapshot.json metadata
        metadata = {
            "version": version,
            "source_migration": migration_name,
            "schema_hash": schema_hash,
            "file_hashes": file_hashes,
            "type_bridge_version": type_bridge_version,
            "type_bridge_core_version": core_version,
        }
        (snapshot_dir / "snapshot.json").write_text(
            json.dumps(metadata, indent=2) + "\n",
            encoding="utf-8",
        )

        logger.info(f"Successfully generated migration snapshot in {snapshot_dir}")
        return snapshot_dir

    except Exception as e:
        # Clean up directory on failure so we don't leave corrupted state
        import shutil

        if snapshot_dir.exists():
            try:
                shutil.rmtree(snapshot_dir)
            except Exception:
                pass
        raise e

get_snapshot_metadata

get_snapshot_metadata(snapshot_dir)

Read metadata from a snapshot.json file.

Source code in type_bridge/migration/snapshots.py

def get_snapshot_metadata(snapshot_dir: Path) -> dict[str, Any] | None:
    """Read metadata from a snapshot.json file."""
    metadata_path = snapshot_dir / "snapshot.json"
    if not metadata_path.exists():
        return None
    try:
        return json.loads(metadata_path.read_text(encoding="utf-8"))
    except Exception as e:
        logger.warning(f"Failed to read snapshot metadata from {metadata_path}: {e}")
        return None

get_snapshot_class_map

get_snapshot_class_map(migrations_dir, version)

Return a mapping from TypeDB type label to class for a snapshot version.

For example, maps 'user' to migrations.snapshots.v0001.entities.User class.

Source code in type_bridge/migration/snapshots.py

def get_snapshot_class_map(migrations_dir: Path, version: str) -> dict[str, type]:
    """Return a mapping from TypeDB type label to class for a snapshot version.

    For example, maps 'user' to migrations.snapshots.v0001.entities.User class.
    """
    import importlib
    import sys

    # Invalidate import caches so newly generated snapshot files are discoverable
    importlib.invalidate_caches()

    app_name = migrations_dir.name
    module_name = f"{app_name}.snapshots.{version}.registry"

    # Clean up sys.modules cache for the migrations package so Python imports it fresh from the new path
    for key in list(sys.modules.keys()):
        if key == app_name or key.startswith(f"{app_name}."):
            try:
                del sys.modules[key]
            except KeyError:
                pass

    parent_path = str(migrations_dir.parent.resolve())
    added_to_path = False
    if parent_path not in sys.path:
        sys.path.insert(0, parent_path)
        added_to_path = True
    try:
        reg_mod = importlib.import_module(module_name)
    except Exception as e:
        logger.warning(f"Could not load registry for snapshot {version}: {e}")
        return {}
    finally:
        if added_to_path:
            try:
                sys.path.remove(parent_path)
            except ValueError:
                pass

    class_map = {}
    if hasattr(reg_mod, "ENTITY_MAP"):
        class_map.update(reg_mod.ENTITY_MAP)
    if hasattr(reg_mod, "RELATION_MAP"):
        class_map.update(reg_mod.RELATION_MAP)
    if hasattr(reg_mod, "ATTRIBUTE_MAP"):
        class_map.update(reg_mod.ATTRIBUTE_MAP)
    return class_map