Skip to content

graflo.architecture.contract.manifest

Graph manifest model for complete ingestion contracts.

Attributes

Classes

GraphManifest

Bases: ConfigBaseModel

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

Source code in graflo/architecture/contract/manifest.py 
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
86
87
88
89
90
91
92
93
94
95
96
97
98
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.",
    )
    metadata: ManifestMetadata | None = PydanticField(
        default=None,
        description=(
            "Manifest-level identity: name, description and provenance. "
            "Excluded from this manifest's own content hash — a content "
            "address that covered its name or its lineage would not be "
            "path-independent, and two routes to the same world model could "
            "never be recognised as equal."
        ),
    )

    @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,
                target_db_flavor=self.graph_schema.db_profile.db_flavor,
            )

    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

Attributes

bindings = PydanticField(default=None, description='Bindings mapping resources to concrete data sources.') class-attribute instance-attribute
graph_schema = PydanticField(default=None, description='Logical graph schema contract.', validation_alias=AliasChoices('schema', 'graph_schema'), serialization_alias='schema') class-attribute instance-attribute
ingestion_model = PydanticField(default=None, description='Ingestion resources and transforms.') class-attribute instance-attribute
metadata = PydanticField(default=None, description="Manifest-level identity: name, description and provenance. Excluded from this manifest's own content hash — a content address that covered its name or its lineage would not be path-independent, and two routes to the same world model could never be recognised as equal.") class-attribute instance-attribute
model_config = ConfigDict(populate_by_name=True) class-attribute instance-attribute

Methods:

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 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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,
            target_db_flavor=self.graph_schema.db_profile.db_flavor,
        )
from_config(data) classmethod

Build a manifest from a Python mapping payload.

Source code in graflo/architecture/contract/manifest.py 
48
49
50
51
@classmethod
def from_config(cls, data: dict[str, Any]) -> GraphManifest:
    """Build a manifest from a Python mapping payload."""
    return cls.from_dict(data)
require_bindings()
Source code in graflo/architecture/contract/manifest.py 
95
96
97
98
def require_bindings(self) -> Bindings:
    if self.bindings is None:
        raise ValueError("GraphManifest is missing required 'bindings' block.")
    return self.bindings
require_ingestion_model()
Source code in graflo/architecture/contract/manifest.py 
88
89
90
91
92
93
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
require_schema()
Source code in graflo/architecture/contract/manifest.py 
83
84
85
86
def require_schema(self) -> Schema:
    if self.graph_schema is None:
        raise ValueError("GraphManifest is missing required 'schema' block.")
    return self.graph_schema