Skip to content

graflo.architecture.contract.manifest

Graph manifest model for complete ingestion contracts.

GraphManifest

Bases: ConfigBaseModel

Canonical config contract for graph schema, ingestion, and bindings.

Source code in graflo/architecture/contract/manifest.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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
class GraphManifest(ConfigBaseModel):
    """Canonical config contract for graph schema, ingestion, and bindings."""

    model_config = ConfigDict(populate_by_name=True)

    graph_schema: Schema | None = PydanticField(
        default=None,
        description="Logical graph schema contract.",
        validation_alias=AliasChoices("schema", "graph_schema"),
        serialization_alias="schema",
    )
    ingestion_model: IngestionModel | None = PydanticField(
        default=None,
        description="Ingestion resources and transforms.",
    )
    bindings: Bindings | None = PydanticField(
        default=None,
        description="Bindings mapping resources to concrete data sources.",
    )

    @classmethod
    def from_config(cls, data: dict[str, Any]) -> "GraphManifest":
        """Build a manifest from a Python mapping payload."""
        return cls.from_dict(data)

    @model_validator(mode="after")
    def _validate_manifest(self) -> "GraphManifest":
        if (
            self.graph_schema is None
            and self.ingestion_model is None
            and self.bindings is None
        ):
            raise ValueError(
                "GraphManifest requires at least one block: "
                "schema, ingestion_model, or bindings."
            )
        return self

    def finish_init(
        self,
        *,
        strict_references: bool = False,
        dynamic_edge_feedback: bool = False,
    ) -> None:
        """Initialize model internals and cross-block runtime links."""
        if self.graph_schema is not None:
            self.graph_schema.finish_init()
        if self.graph_schema is not None and self.ingestion_model is not None:
            self.ingestion_model.finish_init(
                self.graph_schema.core_schema,
                strict_references=strict_references,
                dynamic_edge_feedback=dynamic_edge_feedback,
            )

    def require_schema(self) -> Schema:
        if self.graph_schema is None:
            raise ValueError("GraphManifest is missing required 'schema' block.")
        return self.graph_schema

    def require_ingestion_model(self) -> IngestionModel:
        if self.ingestion_model is None:
            raise ValueError(
                "GraphManifest is missing required 'ingestion_model' block."
            )
        return self.ingestion_model

    def require_bindings(self) -> Bindings:
        if self.bindings is None:
            raise ValueError("GraphManifest is missing required 'bindings' block.")
        return self.bindings

finish_init(*, strict_references=False, dynamic_edge_feedback=False)

Initialize model internals and cross-block runtime links.

Source code in graflo/architecture/contract/manifest.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def finish_init(
    self,
    *,
    strict_references: bool = False,
    dynamic_edge_feedback: bool = False,
) -> None:
    """Initialize model internals and cross-block runtime links."""
    if self.graph_schema is not None:
        self.graph_schema.finish_init()
    if self.graph_schema is not None and self.ingestion_model is not None:
        self.ingestion_model.finish_init(
            self.graph_schema.core_schema,
            strict_references=strict_references,
            dynamic_edge_feedback=dynamic_edge_feedback,
        )

from_config(data) classmethod

Build a manifest from a Python mapping payload.

Source code in graflo/architecture/contract/manifest.py 
36
37
38
39
@classmethod
def from_config(cls, data: dict[str, Any]) -> "GraphManifest":
    """Build a manifest from a Python mapping payload."""
    return cls.from_dict(data)