Schema & Data Model

Purple8 Hyper Graph operates on a data-first model — you can write data immediately without defining a schema. The schema is optional, and when you do define one, it can be discovered from existing data rather than declared upfront.

The spectrum

Purple8 operates across a full spectrum from completely schemaless to strictly enforced:

schemaless          warnings only         strictly enforced
    │                    │                      │
GraphEngine()       SchemaValidator(        SchemaValidator(
                      strict_mode=False)      strict_mode=True)

Pole 1 — Fully schemaless (default)

from purple8_graph import GraphEngine

engine = GraphEngine("./data")

# Any label, any property, any shape — accepted without restriction
engine.add_node("n1", labels=["Person"],   properties={"name": "Alice", "age": 30})
engine.add_node("n2", labels=["Document"], properties={"title": "Q4 Report", "pages": 42})
engine.add_node("n3", labels=["Whatever"], properties={"foo": "bar", "nested": {"a": 1}})

No schema file. No DDL. No migration. Just write.

Pole 2 — Warnings only

from purple8_graph.validation import SchemaValidator, ValidatingGraphEngine, GraphSchema, NodeSchema, PropertySchema, PropertyType

schema = GraphSchema(name="my_schema")
schema.add_node_schema(NodeSchema(
    label="Person",
    properties=[
        PropertySchema(name="name", type=PropertyType.STRING, required=True),
        PropertySchema(name="age",  type=PropertyType.INTEGER),
    ],
))

validator = SchemaValidator(strict_mode=False, allow_extra_properties=True)
validated_engine = ValidatingGraphEngine(engine, validator, schema)

# Validates but doesn't reject — logs warnings for unknown labels or missing required fields
validated_engine.add_node("n4", labels=["UnknownLabel"], properties={"x": 1})

Pole 3 — Strict enforcement

validator_strict = SchemaValidator(strict_mode=True, allow_extra_properties=False)
validated_engine = ValidatingGraphEngine(engine, validator_strict, schema)

# This raises ValidationError — "UnknownLabel" is not in the schema
validated_engine.add_node("n5", labels=["UnknownLabel"], properties={"x": 1})

# This raises ValidationError — "age" is the wrong type
validated_engine.add_node("n6", labels=["Person"], properties={"name": "Bob", "age": "thirty"})

Discovering the schema from existing data

The most important feature of Purple8's schema model: schema can be an output of discovery, not an input.

from purple8_graph.validation import create_schema_from_graph

# Build a knowledge graph from 10,000 documents — no schema upfront
engine = GraphEngine("./data")

# ... add thousands of nodes and edges from LLM extraction ...

# NOW infer the schema from what was actually written
schema = create_schema_from_graph(engine)

print(schema.node_schemas)
# → [Person, Organization, Location, Event, Document, ...]

print(schema.edge_schemas)
# → [WORKS_FOR, LOCATED_IN, ATTENDED, AUTHORED_BY, ...]

# Now enforce it going forward
validator = SchemaValidator(strict_mode=True)
validated_engine = ValidatingGraphEngine(engine, validator, schema)

create_schema_from_graph() scans all nodes and edges, infers property types from observed values, and returns a GraphSchema you can inspect, modify, and enforce.

LLM-inferred schema

For teams who want to define a schema from sample documents before any data exists:

from purple8_graph.genai import SchemaDetector, OpenAIProvider

provider = OpenAIProvider(api_key="...")
detector = SchemaDetector(provider)

# Feed sample documents — LLM infers entities and relationships
schema = detector.detect_schema(sample_documents[:50])

# Returns GraphSchema with NodeSchema + EdgeSchema inferred by the LLM
print(schema.node_schemas)   # → [Person, Organization, Document, ...]

The schema is an output of AI inference — not a prerequisite.

Property types

PropertyType	Python equivalent	Example
STRING	str	"Alice"
INTEGER	int	42
FLOAT	float	3.14
BOOLEAN	bool	True
DATETIME	datetime	datetime(2026, 3, 25)
LIST	list	[1, 2, 3]
DICT	dict	{"key": "value"}

Comparison with schema-first systems

System	Schema model	First write requires schema?	Retroactive schema inference?
Purple8	Data-first, optional strict mode	❌ No	✅ create_schema_from_graph()
Neo4j	Schema-optional (explicit constraints)	❌ No	❌ No
Kùzu	Schemaless	❌ No	❌ No
Spanner Graph	Schema-first (DDL required)	✅ Yes	❌ No
TigerGraph	Schema-first (DDL required)	✅ Yes	❌ No

Why this matters for AI workloads

When you build a knowledge graph from LLM-extracted entities and relationships, the schema is emergent — it's a property of your data that you discover, not something you can define in advance. Spanner Graph and TigerGraph require DDL before the first byte of data. Purple8 lets you write first, understand your data, then optionally enforce a schema.

---

Referential Integrity (v0.40.0)

When an EdgeSchema registers source_labels and/or target_labels, Purple8 enforces these constraints at write time via engine.add_edge().

from purple8_graph.core.models import EdgeSchema, NodeSchema, GraphSchema, PropertySchema, PropertyType

schema = GraphSchema(name="hr_graph")

schema.add_node_schema(NodeSchema(label="Employee"))
schema.add_node_schema(NodeSchema(label="Department"))
schema.add_edge_schema(EdgeSchema(
    edge_type="WORKS_IN",
    source_labels=["Employee"],   # source must have this label
    target_labels=["Department"], # target must have this label
))

engine = GraphEngine("./hr", schema=schema)

engine.add_node("e1", labels=["Employee"],   properties={"name": "Alice"})
engine.add_node("d1", labels=["Department"], properties={"name": "Engineering"})

# ✅ Passes — labels match the schema constraints
engine.add_edge("w1", "WORKS_IN", "e1", "d1")

# ❌ Raises GraphEngineError — "d1" is a Department, not an Employee
engine.add_edge("w2", "WORKS_IN", "d1", "e1")
# → GraphEngineError: Schema validation failed for edge 'w2':
#     Edge 'w2' source node has invalid labels.
#     Expected one of ['Employee'], got ['Department']

Referential integrity is a no-op when no EdgeSchema is registered — existing schemaless graphs are unaffected.

---

Schema Migrations (v0.40.0)

Evolve your GraphSchema without full re-ingest using SchemaMigration and MigrationRunner.

Supported operations

Operation	Description
ADD_PROPERTY	Add a property (with optional default value) to all matching nodes/edges
REMOVE_PROPERTY	Remove a property key from all matching entities
RENAME_PROPERTY	Rename a property key while preserving its value
CHANGE_PROPERTY_TYPE	Coerce a property value via a converter callable
ADD_LABEL	Attach a new label to all nodes carrying the target label
REMOVE_LABEL	Strip a label from all matching nodes

Quick example

from purple8_graph.core.migration import (
    MigrationOperationType,
    MigrationOperation,
    SchemaMigration,
    MigrationRunner,
)

# Define a migration from schema v1.0.0 → v1.1.0
migration = SchemaMigration(
    migration_id="add_email_to_person",
    from_version="1.0.0",
    to_version="1.1.0",
    description="Add optional email property to all Person nodes",
    operations=[
        MigrationOperation(
            op_type=MigrationOperationType.ADD_PROPERTY,
            target_label="Person",
            property_name="email",
            default_value=None,          # existing nodes get None
        ),
        MigrationOperation(
            op_type=MigrationOperationType.RENAME_PROPERTY,
            target_label="Person",
            property_name="full_name",
            new_property_name="name",    # rename full_name → name
        ),
    ],
)

# Apply directly via the engine (version-gated)
result = engine.apply_migration(migration)
print(result)
# → MigrationResult('add_email_to_person': 42 affected, 0 skipped, 0 errors)

print(engine.schema.version)  # → "1.1.0"

Dry run (preview without writing)

result = engine.apply_migration(migration, dry_run=True)
print(f"Would affect {result.total_affected} entities")

Migration registry + multi-step paths

from purple8_graph.core.migration import MigrationRegistry

registry = MigrationRegistry()
registry.register(migration_1_0_to_1_1)
registry.register(migration_1_1_to_1_2)
registry.register(migration_1_2_to_1_3)

engine.register_migration(migration_1_0_to_1_1)
engine.register_migration(migration_1_1_to_1_2)
engine.register_migration(migration_1_2_to_1_3)

# Apply all migrations from current version to target in one call
results = engine.apply_migration_path("1.3.0")
for r in results:
    print(r)

---

Query Cost-Based Planner (v0.40.0)

Purple8's QueryPlanner analyses your query before execution and selects the cheapest physical strategy automatically.

How it works

The planner evaluates three candidate plans:

Strategy	Cost multiplier	When selected
PROPERTY_INDEX_SCAN	×1.0	Equality filter on an indexed property + label filter
LABEL_INDEX_SCAN	×2.0	Label filter present (no indexed property match)
FULL_SCAN	×10.0	Fallback — no usable index

The plan is chosen at get_nodes() call time, not at query compile time. Strategy selection is logged at DEBUG level.

Introspect without executing

from purple8_graph.core.query import Query

q = Query().labels("Person").where("email", "=", "alice@example.com")
plan = engine.get_query_plan(q)

print(plan)
# → QueryPlan(PROPERTY_INDEX_SCAN, cost=4.2, steps=[PROPERTY_INDEX_SCAN → LIMIT])

print(plan.to_dict())
# → {
#     "chosen_strategy": "PROPERTY_INDEX_SCAN",
#     "total_cost": 4.2,
#     "steps": [{"op": "PROPERTY_INDEX_SCAN", "estimated_rows": 4, ...}],
#     "warnings": []
#   }

Ensure an index is present for your hot query paths

# Build a property index so the planner can use PROPERTY_INDEX_SCAN
engine.build_property_index("Person", "email")

# Now repeated queries on email will use the index path
nodes = engine.get_nodes(query=Query().labels("Person").where("email", "=", "alice@example.com"))

TIP

Use engine.get_query_plan(query) in development to verify your query will use an index before deploying to production.