Skip to content

Temporal Knowledge Graph

trails.temporal provides opt-in bitemporal versioning for @node_type instances. Every temporal snapshot lives in its own named graph; temporal metadata (valid_from, valid_until, transaction_time, revision chain) is stored as annotations linked via prov:wasRevisionOf. Non-temporal code is completely unaffected -- all functions degrade gracefully when no temporal data exists.

See ADR-0035 for the full design rationale.

Quick start

from datetime import datetime
from trails.temporal import temporal_save, history, as_of, temporal_diff

# Save a temporal snapshot of a @node_type instance
graph_iri = temporal_save(ctx, note, valid_from=datetime(2026, 1, 1))

# Update and save again -- automatically links via prov:wasRevisionOf
note.title = "Updated title"
graph_iri2 = temporal_save(ctx, note, valid_from=datetime(2026, 6, 1))

# Retrieve full history ordered by valid_from
snapshots = history(ctx, note.id)
for snap in snapshots:
    print(snap.valid_from, snap.fields)

# Point-in-time query: what did Notes look like on 2026-03-01?
from trails.temporal import as_of
notes = as_of(ctx, Note, datetime(2026, 3, 1)).fetch()

# Field-level diff between two points in time
changes = temporal_diff(ctx, note.id, datetime(2026, 1, 1), datetime(2026, 6, 1))
# {"title": ("Original title", "Updated title")}

Key types

Type Description
TemporalMetadata Frozen dataclass: valid_from, valid_until, transaction_time, supersedes
TemporalSnapshot Point-in-time snapshot with graph_iri, node_iri, temporal metadata, and fields dict
TemporalQueryBuilder Returned by as_of() -- call .fetch() to get hydrated instances valid at the given datetime

API

Function Signature Description
temporal_save (ctx, instance, *, valid_from=None, valid_until=None) -> str Save a @node_type instance as a temporal snapshot; returns snapshot graph IRI
history (ctx, iri, *, since=None, until=None) -> list[TemporalSnapshot] Retrieve all snapshots for a node, ordered by valid_from
as_of (ctx, node_type, dt) -> TemporalQueryBuilder Query builder scoped to data valid at dt; falls back to current state if no temporal data
temporal_diff (ctx, iri, t1, t2) -> dict[str, tuple[Any, Any]] Field-level diff between two points in time