Core

Base Pydantic models and conversion logic for domain-specific graph schemas

DomainNode

Base Pydantic model for all domain-specific graph nodes. Provides automatic conversion to the generic GraphNode format used by graph plugins.

Subclasses define typed fields that become GraphNode.properties. The class name becomes the node label by default.

---

DomainNode

def DomainNode(
    data:Any
)->None:

Base Pydantic model for domain-specific graph nodes.

---

DomainNode.get_label

def get_label(
    
)->str: # Node label for the graph (defaults to class name)

Return the node label.

---

DomainNode.to_graph_node

def to_graph_node(
    sources:List=[], # External data references for provenance
)->GraphNode: # Generic GraphNode for storage in graph plugins

Convert this domain model to a generic GraphNode.

Basic Usage

# Define a simple domain node subclass
class Person(DomainNode):
    name: str
    role: Optional[str] = None

# Create instance with auto-generated ID
alice = Person(name="Alice", role="speaker")
print(f"Person: {alice}")
print(f"Label: {alice.get_label()}")

Person: id='a8a100d3-4f7b-4ad6-a476-e0a0d07957a6' name='Alice' role='speaker'
Label: Person

# Convert to GraphNode
graph_node = alice.to_graph_node()
print(f"GraphNode ID: {graph_node.id}")
print(f"GraphNode label: {graph_node.label}")
print(f"GraphNode properties: {graph_node.properties}")

GraphNode ID: a8a100d3-4f7b-4ad6-a476-e0a0d07957a6
GraphNode label: Person
GraphNode properties: {'name': 'Alice', 'role': 'speaker'}

With Source References

# Create a source reference to external data
sample_text = b"Laying Plans Sun Tzu said, The art of war is of vital importance..."
content_hash = SourceRef.compute_hash(sample_text)

source = SourceRef(
    plugin_name="cjm-transcription-plugin-voxtral-hf",
    table_name="transcriptions",
    row_id="job-123",
    content_hash=content_hash,
    segment_slice="timestamp:00:10-00:30"
)

# Convert with provenance tracking
graph_node = alice.to_graph_node(sources=[source])
print(f"Sources: {len(graph_node.sources)}")
print(f"Source ref: {graph_node.sources[0]}")

Sources: 1
Source ref: SourceRef(plugin_name='cjm-transcription-plugin-voxtral-hf', table_name='transcriptions', row_id='job-123', content_hash='sha256:90fc4487461482ec8f7022c3e058f6c026959a444b7deb6cd28dd77190a99de0', segment_slice='timestamp:00:10-00:30')

Name Normalization

The to_graph_node() method automatically normalizes the name property for consistent display. If no name field exists, it falls back to title, label, or truncated text.

# Domain node with 'title' instead of 'name'
class Book(DomainNode):
    title: str
    author: Optional[str] = None

book = Book(title="The Art of War", author="Sun Tzu")
graph_node = book.to_graph_node()

# 'name' is auto-populated from 'title'
print(f"Properties: {graph_node.properties}")
assert graph_node.properties['name'] == "The Art of War"

Properties: {'title': 'The Art of War', 'author': 'Sun Tzu', 'name': 'The Art of War'}

# Domain node with 'text' field (truncated to 50 chars)
class Quote(DomainNode):
    text: str

quote = Quote(text="This is a very long quote that exceeds fifty characters and will be truncated for the name field.")
graph_node = quote.to_graph_node()

print(f"Name (truncated): '{graph_node.properties['name']}'")
assert len(graph_node.properties['name']) == 50

Name (truncated): 'This is a very long quote that exceeds fifty chara'