Skip to content

graflo.architecture.schema.document

Full graph schema document (metadata + core + DB profile).

Classes

Schema

Bases: ConfigBaseModel

Graph schema (A+B): metadata, core schema, and DB profile.

Source code in graflo/architecture/schema/document.py 
 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
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class Schema(ConfigBaseModel):
    """Graph schema (A+B): metadata, core schema, and DB profile."""

    metadata: GraphMetadata = PydanticField(
        ...,
        description="Schema metadata and versioning (name, version).",
    )
    core_schema: CoreSchema = PydanticField(
        ...,
        description="Core schema model (vertices + edges).",
        validation_alias=AliasChoices("core_schema", "graph"),
    )
    db_profile: DatabaseProfile = PydanticField(
        default_factory=DatabaseProfile,
        description=(
            "Database-specific physical profile (secondary indexes, naming, TigerGraph GSQL "
            "DEFAULT overrides via default_property_values, etc.)."
        ),
    )

    @model_validator(mode="after")
    def _init_schema(self) -> Schema:
        self.finish_init()
        return self

    def finish_init(self) -> None:
        from .db_aware import compile_secondary_identity_indexes

        self.core_schema.finish_init()
        compile_secondary_identity_indexes(
            self.core_schema.vertex_config, self.db_profile
        )
        # Normalize away empty index entries left by index or identity removals:
        # they mean the same as no entry but hash differently, which would make
        # a replayed manifest compare unequal to an identically-authored one.
        self.db_profile.prune_empty_vertex_indexes()
        self.db_profile.validate_against_schema(self.core_schema.edge_config)
        self.db_profile.validate_native_inverses(
            self.core_schema.edge_config, self.core_schema.vertex_config.vertex_set
        )

    def effective_namespace(
        self, db_flavor: DBType | None = None, override: str | None = None
    ) -> str:
        """The database / graph / space this schema deploys into on *db_flavor*.

        ``metadata.name`` is a label; this is the identifier. See
        :func:`~graflo.architecture.schema.namespace.resolve_namespace` for the
        precedence (override, ``db_profile.target_namespace``, sanitized name).
        """
        return resolve_namespace(self, db_flavor, override)

    def remove_disconnected_vertices(self) -> set[str]:
        return self.core_schema.remove_disconnected_vertices()

    def resolve_db_aware(self, db_flavor: DBType | None = None) -> SchemaDBAware:
        """Build DB-aware runtime wrappers for *db_flavor*.

        Non-mutating: when *db_flavor* differs from the authored profile, the
        wrappers are built over a copy. This used to assign onto
        ``self.db_profile``, which made the schema's flavor a function of
        whoever resolved it last — fine while resolution only ever happened once
        per ingest, wrong as soon as two readers want two different targets from
        one shared ``Schema``.
        """
        from .db_aware import (
            EdgeConfigDBAware,
            SchemaDBAware,
            VertexConfigDBAware,
        )

        profile = self.db_profile
        if db_flavor is not None and profile.db_flavor != db_flavor:
            profile = profile.model_copy(deep=True)
            profile.db_flavor = db_flavor

        vertex_db = VertexConfigDBAware(self.core_schema.vertex_config, profile)
        edge_db = EdgeConfigDBAware(self.core_schema.edge_config, vertex_db, profile)
        vertex_db.compile_secondary_identity_indexes()
        edge_db.compile_identity_indexes()
        return SchemaDBAware(
            vertex_config=vertex_db,
            edge_config=edge_db,
            db_profile=profile,
        )

    @staticmethod
    def _slug_filename_token(token: str) -> str:
        """Normalize arbitrary token into filename-safe slug."""
        return slugify(token, fallback="schema")

    def default_dump_filename(self) -> str:
        """Return default schema dump filename: <name>-<version>.yaml."""
        schema_name = self._slug_filename_token(self.metadata.name)
        version = (
            self.metadata.version
            if self.metadata.version is not None
            else "unversioned"
        )
        schema_version = self._slug_filename_token(version)
        return f"{schema_name}-{schema_version}.yaml"

    def dump(
        self,
        path: str | pathlib.Path | None = None,
        *,
        exclude_defaults: bool = True,
    ) -> pathlib.Path:
        """Dump schema YAML to path, excluding defaults by default.

        If path is omitted, writes into current working directory using
        `<schema_name>-<version>.yaml`.
        """
        if path is None:
            target_path = pathlib.Path.cwd() / self.default_dump_filename()
        else:
            target_path = pathlib.Path(path)
            if target_path.is_dir():
                target_path = target_path / self.default_dump_filename()

        if exclude_defaults:
            payload = self.to_minimal_canonical_dict()
        else:
            payload = self.to_dict(skip_defaults=False)
        target_path.parent.mkdir(parents=True, exist_ok=True)
        target_path.write_text(
            yaml.safe_dump(
                payload,
                default_flow_style=False,
                sort_keys=False,
            ),
            encoding="utf-8",
        )
        return target_path

Attributes

core_schema = PydanticField(..., description='Core schema model (vertices + edges).', validation_alias=AliasChoices('core_schema', 'graph')) class-attribute instance-attribute
db_profile = PydanticField(default_factory=DatabaseProfile, description='Database-specific physical profile (secondary indexes, naming, TigerGraph GSQL DEFAULT overrides via default_property_values, etc.).') class-attribute instance-attribute
metadata = PydanticField(..., description='Schema metadata and versioning (name, version).') class-attribute instance-attribute

Methods:

default_dump_filename()

Return default schema dump filename: -.yaml.

Source code in graflo/architecture/schema/document.py 
115
116
117
118
119
120
121
122
123
124
def default_dump_filename(self) -> str:
    """Return default schema dump filename: <name>-<version>.yaml."""
    schema_name = self._slug_filename_token(self.metadata.name)
    version = (
        self.metadata.version
        if self.metadata.version is not None
        else "unversioned"
    )
    schema_version = self._slug_filename_token(version)
    return f"{schema_name}-{schema_version}.yaml"
dump(path=None, *, exclude_defaults=True)

Dump schema YAML to path, excluding defaults by default.

If path is omitted, writes into current working directory using <schema_name>-<version>.yaml.

Source code in graflo/architecture/schema/document.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def dump(
    self,
    path: str | pathlib.Path | None = None,
    *,
    exclude_defaults: bool = True,
) -> pathlib.Path:
    """Dump schema YAML to path, excluding defaults by default.

    If path is omitted, writes into current working directory using
    `<schema_name>-<version>.yaml`.
    """
    if path is None:
        target_path = pathlib.Path.cwd() / self.default_dump_filename()
    else:
        target_path = pathlib.Path(path)
        if target_path.is_dir():
            target_path = target_path / self.default_dump_filename()

    if exclude_defaults:
        payload = self.to_minimal_canonical_dict()
    else:
        payload = self.to_dict(skip_defaults=False)
    target_path.parent.mkdir(parents=True, exist_ok=True)
    target_path.write_text(
        yaml.safe_dump(
            payload,
            default_flow_style=False,
            sort_keys=False,
        ),
        encoding="utf-8",
    )
    return target_path
effective_namespace(db_flavor=None, override=None)

The database / graph / space this schema deploys into on db_flavor.

metadata.name is a label; this is the identifier. See :func:~graflo.architecture.schema.namespace.resolve_namespace for the precedence (override, db_profile.target_namespace, sanitized name).

Source code in graflo/architecture/schema/document.py 
65
66
67
68
69
70
71
72
73
74
def effective_namespace(
    self, db_flavor: DBType | None = None, override: str | None = None
) -> str:
    """The database / graph / space this schema deploys into on *db_flavor*.

    ``metadata.name`` is a label; this is the identifier. See
    :func:`~graflo.architecture.schema.namespace.resolve_namespace` for the
    precedence (override, ``db_profile.target_namespace``, sanitized name).
    """
    return resolve_namespace(self, db_flavor, override)
finish_init()
Source code in graflo/architecture/schema/document.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def finish_init(self) -> None:
    from .db_aware import compile_secondary_identity_indexes

    self.core_schema.finish_init()
    compile_secondary_identity_indexes(
        self.core_schema.vertex_config, self.db_profile
    )
    # Normalize away empty index entries left by index or identity removals:
    # they mean the same as no entry but hash differently, which would make
    # a replayed manifest compare unequal to an identically-authored one.
    self.db_profile.prune_empty_vertex_indexes()
    self.db_profile.validate_against_schema(self.core_schema.edge_config)
    self.db_profile.validate_native_inverses(
        self.core_schema.edge_config, self.core_schema.vertex_config.vertex_set
    )
remove_disconnected_vertices()
Source code in graflo/architecture/schema/document.py 
76
77
def remove_disconnected_vertices(self) -> set[str]:
    return self.core_schema.remove_disconnected_vertices()
resolve_db_aware(db_flavor=None)

Build DB-aware runtime wrappers for db_flavor.

Non-mutating: when db_flavor differs from the authored profile, the wrappers are built over a copy. This used to assign onto self.db_profile, which made the schema's flavor a function of whoever resolved it last — fine while resolution only ever happened once per ingest, wrong as soon as two readers want two different targets from one shared Schema.

Source code in graflo/architecture/schema/document.py 
 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
def resolve_db_aware(self, db_flavor: DBType | None = None) -> SchemaDBAware:
    """Build DB-aware runtime wrappers for *db_flavor*.

    Non-mutating: when *db_flavor* differs from the authored profile, the
    wrappers are built over a copy. This used to assign onto
    ``self.db_profile``, which made the schema's flavor a function of
    whoever resolved it last — fine while resolution only ever happened once
    per ingest, wrong as soon as two readers want two different targets from
    one shared ``Schema``.
    """
    from .db_aware import (
        EdgeConfigDBAware,
        SchemaDBAware,
        VertexConfigDBAware,
    )

    profile = self.db_profile
    if db_flavor is not None and profile.db_flavor != db_flavor:
        profile = profile.model_copy(deep=True)
        profile.db_flavor = db_flavor

    vertex_db = VertexConfigDBAware(self.core_schema.vertex_config, profile)
    edge_db = EdgeConfigDBAware(self.core_schema.edge_config, vertex_db, profile)
    vertex_db.compile_secondary_identity_indexes()
    edge_db.compile_identity_indexes()
    return SchemaDBAware(
        vertex_config=vertex_db,
        edge_config=edge_db,
        db_profile=profile,
    )

Functions: