kapps_ogm.node.core#

Core Node class for managing RDF-backed entity instances.

class kapps_ogm.node.core.Node(*, id: str | IRI | BNode | None = None, class_spec: 'ClassSpec' | None = None, data: Dict[IRILike, List[Any]] | None = None, instance: T | None = None, ogm: OGM | None = None)[source]#

Bases: object

Lightweight runtime handle for an RDF-backed instance.

A Node represents an entity identified by an IRI and may carry: - a ClassSpec describing its schema, - raw loaded data, - a materialized Pydantic instance.

It coordinates the entity lifecycle, including lazy loading and materialization, without embedding persistence logic itself.

property is_materialized: bool#

Check if the node has a materialized Pydantic instance.

property has_data: bool#

Check if the node has loaded data.

property data: Dict[IRI, List[Any]] | None#

Get the raw instance data for this node.

Returns:

Raw instance data if loaded, else None.

Return type:

Optional[Dict[IRI, List[Any]]]

classmethod from_sanitize_data(ogm: OGM, data: Dict[IRILike, List[Any]], known_nodes: Dict[int, 'Node'] | None = None) → Node[source]#
property class_spec: 'ClassSpec' | None#

Get the ClassSpec associated with this node.

materialize(*, reload: bool = False) → BaseModel[source]#

Return a validated Pydantic model instance for this node.

Lazily materializes the nodes loaded data into a Pydantic model on first access and caches the result. Subsequent calls return the cached instance without reprocessing.

If data has not yet been loaded, it is loaded automatically before materialization.

If reload is True, any cached instance is discarded. The nodes data is reloaded from the database and a new instance is materialized.

Returns:

A validated Pydantic model corresponding to the nodes ClassSpec. The instance is cached on the node.

Return type:

BaseModel

Raises:

  • RuntimeError – If no OGM is attached to load data.

  • ValidationError – If the loaded data violates the ClassSpec constraints.

Notes

  • This is the preferred way to access node data in application code.

  • Modifications to the returned instance must be persisted explicitly via the OGM.

diff(other: Node) → Tuple[Tuple[Tuple[IRI | BNode, IRI, IRI | BNode | Literal]], Tuple[Tuple[IRI | BNode, IRI, IRI | BNode | Literal]]][source]#

Compute the triples to update this node to match new_node.

Parameters:

other (Node) – The target node state to commit to.

Returns:

A tuple containing two tuples:

  • old triples to remove

  • new triples to add

Return type:

Tuple[Tuple[Triple], Tuple[Triple]]

Raises:

NotImplementedError – If other is not a Node instance.

to_triples() → set[Triple]#

Serialize the nodes current instance into RDF triples for persistence.

Uses the in-memory materialized instance and cached data. Updates in memory will be reflected in the serialized triples.

Parameters:

node – The Node instance to serialize.

Returns:

RDF triples representing this node, including nested objects.

Return type:

set[Triple]

Raises:

RuntimeError – If the node has no ClassSpec or IRI.

Notes

  • Intended for persisting the current instance state.

  • Always uses the cached instance; does not refresh from the database.

to_json_ld(context: Dict[str, str] | None = None) → Dict[str, Any]#

Serialize the nodes instance into JSON-LD format.

Converts RDF triples to JSON-LD with proper handling of IRIs, blank nodes, and literals. Blank nodes are inlined recursively for cleaner output.

Parameters:

  • node – The Node instance to serialize.

  • context – Optional @context dictionary mapping prefixes to namespace URIs. If provided, URIs are compacted using these prefixes.

Returns:

JSON-LD document with @context and @graph keys.

Main subject appears first in @graph, followed by other named nodes. Blank nodes are inlined as nested objects.

Return type:

Dict[str, Any]

Notes

  • Calls to_triples() internally to obtain RDF representation

  • Literal datatypes are preserved as native Python types

  • Multiple property values are represented as JSON arrays

  • Context prefixes are registered globally in IRI.PREFIXES

extract_property_chains() → list[list[IRI | str]]#

Extract property chains from nested node data.

Reconstructs full IRIs from lined keys using hybrid lookup: 1. Direct IRI construction (already full IRI) 2. _iri_fields mapping (top-level properties from ClassSpec) 3. Decode from lined format (nested properties) 4. Fallback to string if all fail

Returns:

Property chains from root to each terminal value,

with all keys normalized to full IRIs where possible.

Return type:

list[list[IRI | str]]

log_data_debug() → None[source]#

Pretty print data for debugging.