Skip to content

graflo.migrate.io

I/O utilities for migration workflows.

Attributes

__all__ = ['full_hash', 'graph_hash', 'ingestion_hash', 'load_ingestion_model', 'load_manifest', 'load_schema', 'manifest_hash', 'plan_to_json_serializable', 'schema_hash', 'stable_hash'] module-attribute

Classes

Functions:

full_hash(schema, ingestion_model, bindings)

Stable hash over a merged deployment object.

Source code in graflo/architecture/evolution/hashing.py 
58
59
60
61
62
63
64
65
66
67
68
69
def full_hash(schema: Any, ingestion_model: Any, bindings: Any) -> str:
    """Stable hash over a merged deployment object."""
    payload = {
        "schema": _schema_payload(schema),
        "ingestion": canonical_payload(ingestion_model),
        "bindings": (
            canonical_payload(bindings)
            if hasattr(bindings, "to_minimal_canonical_dict")
            else (bindings.to_dict() if hasattr(bindings, "to_dict") else bindings)
        ),
    }
    return stable_hash(payload)

graph_hash(schema)

Stable hash over the logical graph model only.

Source code in graflo/architecture/evolution/hashing.py 
33
34
35
def graph_hash(schema: Any) -> str:
    """Stable hash over the logical graph model only."""
    return stable_hash(canonical_payload(schema.core_schema))

ingestion_hash(ingestion_model)

Stable hash over the ingestion model (resources + transforms).

Source code in graflo/architecture/evolution/hashing.py 
53
54
55
def ingestion_hash(ingestion_model: Any) -> str:
    """Stable hash over the ingestion model (resources + transforms)."""
    return stable_hash(canonical_payload(ingestion_model))

load_ingestion_model(path, schema=None)

Load ingestion block from a manifest path.

Source code in graflo/migrate/io.py 
48
49
50
51
52
53
54
55
56
57
58
59
def load_ingestion_model(
    path: str | Path, schema: Schema | None = None
) -> IngestionModel:
    """Load ingestion block from a manifest path."""
    manifest = load_manifest(path)
    ingestion_model = manifest.require_ingestion_model()
    if schema is not None:
        ingestion_model.finish_init(
            schema.core_schema,
            target_db_flavor=schema.db_profile.db_flavor,
        )
    return ingestion_model

load_manifest(path)

Load and initialize graph manifest from YAML path.

Source code in graflo/migrate/io.py 
36
37
38
39
40
def load_manifest(path: str | Path) -> GraphManifest:
    """Load and initialize graph manifest from YAML path."""
    manifest = GraphManifest.from_config(FileHandle.load(path))
    manifest.finish_init()
    return manifest

load_schema(path)

Load schema block from a manifest path.

Source code in graflo/migrate/io.py 
43
44
45
def load_schema(path: str | Path) -> Schema:
    """Load schema block from a manifest path."""
    return load_manifest(path).require_schema()

manifest_hash(manifest)

Stable hash over all three manifest blocks.

This is the identity a revision chain verifies against: replaying a chain from its base must reproduce the recorded hash at every step.

Source code in graflo/architecture/evolution/hashing.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def manifest_hash(manifest: Any) -> str:
    """Stable hash over all three manifest blocks.

    This is the identity a revision chain verifies against: replaying a chain
    from its base must reproduce the recorded hash at every step.
    """
    payload = {
        "schema": _schema_payload(manifest.graph_schema),
        "ingestion_model": (
            canonical_payload(manifest.ingestion_model)
            if manifest.ingestion_model is not None
            else None
        ),
        "bindings": (
            canonical_payload(manifest.bindings)
            if manifest.bindings is not None
            else None
        ),
    }
    return stable_hash(payload)

plan_to_json_serializable(plan)

Convert pydantic plan-like object to JSON payload.

Source code in graflo/migrate/io.py 
68
69
70
71
72
73
74
def plan_to_json_serializable(plan: Any) -> dict[str, Any]:
    """Convert pydantic plan-like object to JSON payload."""
    if hasattr(plan, "model_dump"):
        return plan.model_dump()
    if hasattr(plan, "to_dict"):
        return plan.to_dict()
    raise TypeError(f"Unsupported plan object type: {type(plan)}")

schema_hash(schema)

Stable hash over the schema deployment contract (graph + DB profile).

metadata is deliberately excluded: name, semver and description are how a schema is labelled, not what it is, and a content address that moves when the version bumps cannot recognise that two versions carry identical content. :func:manifest_hash excludes it for the same reason.

Source code in graflo/architecture/evolution/hashing.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
def schema_hash(schema: Any) -> str:
    """Stable hash over the schema deployment contract (graph + DB profile).

    ``metadata`` is deliberately excluded: name, semver and description are how
    a schema is *labelled*, not what it *is*, and a content address that moves
    when the version bumps cannot recognise that two versions carry identical
    content. :func:`manifest_hash` excludes it for the same reason.
    """
    payload = {
        "core_schema": canonical_payload(schema.core_schema),
        "db_profile": canonical_payload(schema.db_profile),
    }
    return stable_hash(payload)

stable_hash(payload_obj)

SHA-256 over a canonical JSON rendering of payload_obj.

The canonicalization version is part of the hashed bytes, not a wrapper around them, so it cannot be stripped by a caller that re-serializes.

Source code in graflo/architecture/evolution/hashing.py 
24
25
26
27
28
29
30
def stable_hash(payload_obj: Any) -> str:
    """SHA-256 over a canonical JSON rendering of *payload_obj*.

    The canonicalization version is part of the hashed bytes, not a wrapper
    around them, so it cannot be stripped by a caller that re-serializes.
    """
    return suthing.stable_hash({"canon": CANON_VERSION, "payload": payload_obj})