RiC-O — Graph Model Demo
This notebook demonstrates how to model a RiC-O (Records in Contexts — Ontology) graph using the DRM tools. RiC-O is the ICA ontology for archival description.
Key concepts:
Thingis the single strong (root) node in RiC-OAll other entities (
Agent,Person,CorporateBody,Date,Place,Event,RecordResource, etc.) areWeakNodesubclassescreate_group()atomically creates aThingwith all its weak childrenget_subdocuments()retrieves all weak descendants of a strong nodeload_ric_o_naf()loads real RiC-O data from the National Archives of France example
Steps:
Connect to Neo4j DEV
Load real RiC-O data from the National Archives of France (NAF) example
Initialize propagation properties
Query subdocuments
Inspect the hierarchy
Summary statistics
Prerequisites
A running Neo4j instance (DEV target)
Environment variables:
NEO4J_DEV_URL,NEO4J_DEV_USER,NEO4J_DEV_PASSWORDDefault connection:
bolt://localhost:7687/neo4j/neo4j2026
Connection configuration
The notebook reads connection parameters from environment variables with sensible defaults. Set these before running:
export NEO4J_DEV_URL=bolt://your-host:7687
export NEO4J_DEV_USER=neo4j
export NEO4J_DEV_PASSWORD=your-password
[1]:
import importlib.util
# Check if drm is installed
package_to_check = 'drm'
spec = importlib.util.find_spec(package_to_check)
if spec is None:
print(f'⚠️ {package_to_check} not installed. Installing...')
%pip install -q --upgrade drm-tools
print("✅ Installation complete.")
else:
print(f'✅ {package_to_check} already installed. Skipping.')
✅ drm already installed. Skipping.
[2]:
import os
import sys
from pathlib import Path
# Read .env file manually (no python-dotenv dependency)
def read_env(env_path):
"""Read a .env file and return a dict of key=value pairs."""
env_vars = {}
with open(env_path, 'r') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
if '=' in line:
key, value = line.split('=', 1)
env_vars[key.strip()] = value.strip()
return env_vars
# Find .env in repo root
_env_path = Path.cwd() / ".env"
if not _env_path.exists():
for parent in Path.cwd().parents:
candidate = parent / ".env"
if candidate.exists():
_env_path = candidate
break
if _env_path.exists():
_env_vars = read_env(_env_path)
for key, value in _env_vars.items():
os.environ.setdefault(key, value)
print(f"✅ Loaded .env from {_env_path}")
print(f" Found {len(_env_vars)} variables")
else:
print("⚠️ .env file not found. Set NEO4J_DEV_PASSWORD environment variable manually.")
# Add drm package to path for development
_repo_root = Path.cwd()
if not (_repo_root / "drm").exists():
for parent in Path.cwd().parents:
if (parent / "drm").exists():
_repo_root = parent
break
# Add repo root to sys.path so local drm loads before PyPI version
if str(_repo_root) not in sys.path:
sys.path.insert(0, str(_repo_root))
print(f"✅ Added {_repo_root} to sys.path")
from drm import Neo4jGraph
from drm.rico_entities import (
Thing,
Agent, Person, CorporateBody, AgentName, Appellation,
Date, Place, PlaceName, Event, Activity,
RecordResource, Instantiation, Title,
Group, Name,
)
def neo4j_config(default_target="DEV"):
"""Build Neo4j connection config from environment variables."""
target = os.environ.get("NEO4J_TARGET", default_target).upper()
prefix = f"NEO4J_{target}_"
return {
"target": target,
"url": os.environ.get(f"{prefix}URL", os.environ.get("NEO4J_URL", "bolt://localhost:7687")),
"user": os.environ.get(f"{prefix}USER", os.environ.get("NEO4J_USER", "neo4j")),
"password": os.environ.get(f"{prefix}PASSWORD", os.environ.get("NEO4J_PASSWORD", "secret")),
"database": os.environ.get(f"{prefix}DATABASE", os.environ.get("NEO4J_DATABASE", "neo4j")),
}
def section(title):
"""Print a section header."""
print(f"\n{'=' * 60}")
print(f" {title}")
print(f"{'=' * 60}")
def sub_section(title):
"""Print a subsection header."""
print(f"\n--- {title} ---")
print("✅ All imports successful.")
✅ Loaded .env from /Users/oriol/Desenvolupament/drm-tools/.env
Found 5 variables
✅ All imports successful.
Step 1: Connect to Neo4j
Connect to the DEV Neo4j instance and verify the connection.
[3]:
cfg = neo4j_config()
print(f"Target: {cfg['target']}")
print(f"URL: {cfg['url']}")
print(f"User: {cfg['user']}")
print(f"Database: {cfg['database']}")
graph = Neo4jGraph(
url=cfg["url"],
user=cfg["user"],
password=cfg["password"],
database=cfg["database"],
)
# Verify connection
result = graph.query("RETURN 1 AS test")
if result and result[0].get("test") == 1:
print("\n✅ Connection verified.")
else:
print("\n⚠️ Connection returned unexpected result.")
# Check current state
count_result = graph.query("MATCH (n) RETURN count(n) AS total")
total = count_result[0]["total"] if count_result else 0
print(f"Nodes in database: {total}")
Target: DEV
URL: bolt://localhost:7687
User: neo4j
Database: neo4j
✅ Connection verified.
Nodes in database: 99
Step 1.5: Clear existing data
Delete all nodes and relationships before loading fresh RiC-O data.
[4]:
# Clear all existing data from the database
graph.query("MATCH (n) DETACH DELETE n")
print("✅ All existing nodes and relationships deleted.")
# Verify database is empty
count_result = graph.query("MATCH (n) RETURN count(n) AS total")
total = count_result[0]["total"] if count_result else 0
print(f"Nodes remaining: {total}")
✅ All existing nodes and relationships deleted.
Nodes remaining: 0
Step 2: Load real RiC-O data from the National Archives of France
The DRM tools include a loader that downloads and parses RiC-O RDF/XML data from the ICA-EGAD/RiC-O repository.
What gets loaded:
RDF Type |
DRM Entity |
Parent |
|---|---|---|
|
|
— (strong node) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Hierarchy example:
Thing (strong)
├── Agent (WeakNode)
│ ├── AgentName (WeakNode of Agent)
│ └── Instantiation (WeakNode of Agent)
└── RecordResource (WeakNode)
└── Instantiation (WeakNode of RecordResource)
The loader downloads files from GitHub, parses the RDF/XML, and inserts entities into the graph with proper parent-child relationships.
[5]:
from drm.exemples import load_ric_o_naf
section("Loading RiC-O data from National Archives of France")
# Load real RiC-O data
# limit: total entities to load (soft limit)
# max_agents: number of agent files to load
# max_records: number of record resource files to load
stats = load_ric_o_naf(
graph,
limit=50,
max_agents=10,
max_records=3,
)
print(f"\n📊 Load statistics:")
print(f" Agents loaded: {stats['agents']}")
print(f" Records loaded: {stats['records']}")
print(f" Things created: {stats['things']}")
print(f" Child entities: {stats.get('children', '?')}")
print(f" Total entities: {stats['total']}")
============================================================
Loading RiC-O data from National Archives of France
============================================================
---------------------------------------------------------------------------
RuntimeError Traceback (most recent call last)
Cell In[5], line 9
5 # Load real RiC-O data
6 # limit: total entities to load (soft limit)
7 # max_agents: number of agent files to load
8 # max_records: number of record resource files to load
----> 9 stats = load_ric_o_naf(
10 graph,
11 limit=50,
12 max_agents=10,
File ~/Desenvolupament/drm-tools/drm/exemples/load_ric_o_naf.py:784, in load_ric_o_naf(graph, limit, max_agents, max_records)
782 for node in weak_nodes:
783 if hasattr(node, "_parent") and node._parent is not None:
--> 784 graph.insertNode(node, insert_parent=True)
786 # ── 4. Compute final stats from entity_map ────────────────────
787 agent_labels = {"Agent", "CorporateBody", "Person", "Family"}
File ~/Desenvolupament/drm-tools/drm/neo4j_graph.py:142, in Neo4jGraph.insertNode(self, node, insert_parent, update, replace, **kwargs)
134 if has_parent:
135 self.insertNode(
136 node["parent"],
137 insert_parent=insert_parent,
138 update=True,
139 replace=False,
140 )
--> 142 id = self._insertNode(node, update=update, replace=replace)
144 if has_dependencies:
145 deps = node["dependencies"]
File ~/Desenvolupament/drm-tools/drm/neo4j_graph.py:1161, in Neo4jGraph._insertNode(self, node, update, replace)
1156 if not node["parent"]["pk"]["pk"][ppk] == node["pk"]["pk"][ppk]:
1157 raise RuntimeError(
1158 "ADGT Exception: Integrity Constraint Violated. Child node keys does not reference proper parent keys"
1159 )
-> 1161 self.insertRelation(
1162 WeakRelation(
1163 node["parent"],
1164 node,
1165 node["parent_relation"],
1166 propagate=True,
1167 ),
1168 update=True,
1169 replace=False,
1170 )
1172 except ConstraintError as err:
1173 warnings.warn(f"[XPP Message]: {err.message}")
File ~/Desenvolupament/drm-tools/drm/neo4j_graph.py:419, in Neo4jGraph.insertRelation(self, rel, update, replace, **kwargs)
414 raise RuntimeError(
415 f"FK violation: src node with pk={rel['src'].get('pk')} "
416 f"and main_label={rel['src'].get('main_label')} does not exist."
417 )
418 if not dst_exists:
--> 419 raise RuntimeError(
420 f"FK violation: dst node with pk={rel['dst'].get('pk')} "
421 f"and main_label={rel['dst'].get('main_label')} does not exist."
422 )
423 if update:
424 # return self._session.write_transaction(self._update_relation,rel )
425 id = self._update_relation(self._tx, rel)
RuntimeError: FK violation: dst node with pk={'identifier': 'thing-rr-recordResource-recordResource-top-003500', 'recordResourceId': 'recordResource-recordResource-top-003500', 'instantiationId': 'instantiation-4598756304'} and main_label=Instantiation does not exist.
Step 3: Initialize propagation properties
Scan the graph and initialize _propagate flags on edges between parent and child nodes. This enables cascade delete behavior.
[ ]:
# Check state before init
sub_section("Before init_propagation()")
result = graph.query(
"MATCH (n) RETURN count(n) AS total, "
"count(n.is_weak) AS weak_count, "
"count(n._weak_init_done) AS done_count"
)
row = result[0] if result else {}
print(f" Total nodes: {row.get('total', 0)}")
print(f" Nodes with is_weak: {row.get('weak_count', 0)}")
print(f" Nodes with _weak_init_done: {row.get('done_count', 0)}")
# Run initialization
print("\nRunning init_propagation()...")
init_result = graph.init_propagation()
print(f"Result: {init_result}")
# Check state after init
sub_section("After init_propagation()")
result = graph.query(
"MATCH (n) RETURN count(n) AS total, "
"count(n.is_weak) AS weak_count, "
"count(n._weak_init_done) AS done_count"
)
row = result[0] if result else {}
print(f" Total nodes: {row.get('total', 0)}")
print(f" Nodes with is_weak: {row.get('weak_count', 0)}")
print(f" Nodes with _weak_init_done: {row.get('done_count', 0)}")
# Show propagation edges
result = graph.query(
"MATCH ()-[r]->() WHERE r._propagate = TRUE RETURN count(r) AS c"
)
prop_count = result[0]["c"] if result else 0
print(f" Edges with _propagate: {prop_count}")
Step 4: Inspect the loaded graph structure
Query the database to see what entities were loaded and how they are organized.
[ ]:
# Show nodes by label
sub_section("Nodes by label")
result = graph.query(
"MATCH (n) RETURN labels(n)[0] AS label, count(n) AS cnt "
"ORDER BY cnt DESC"
)
for row in result:
print(f" {row['label']}: {row['cnt']}")
# Show relations by type
sub_section("Relations by type")
result = graph.query(
"MATCH ()-[r]->() RETURN type(r) AS rel_type, count(r) AS cnt "
"ORDER BY cnt DESC"
)
for row in result:
print(f" {row['rel_type']}: {row['cnt']}")
[ ]:
# Show all Things
sub_section("All Things (strong nodes)")
result = graph.query(
"MATCH (t:Thing) RETURN t.identifier AS id, t.name AS name "
"ORDER BY id"
)
for row in result:
print(f" {row['id']}: {row['name']}")
Step 5: Get subdocuments
Use get_subdocuments() to retrieve all weak descendants of each Thing. This traverses only edges with _propagate=True.
[ ]:
# Show subdocuments for each Thing
result = graph.query(
"MATCH (t:Thing) RETURN t.identifier AS id ORDER BY id"
)
for row in result:
thing_id = row['id']
thing_node = Thing(pk={"identifier": thing_id})
subdocs = graph.get_subdocuments(thing_node)
sub_section(f"Subdocuments of Thing({thing_id})")
print(f" Found {len(subdocs)} subdocuments:")
# Group by label
by_label = {}
for sd in subdocs:
label = sd['label']
if label not in by_label:
by_label[label] = []
by_label[label].append(sd)
for label in sorted(by_label.keys()):
items = by_label[label]
for item in items[:5]: # Show max 5 per label
name = item['pk'].get('fullName', item['pk'].get('name', item['pk'].get('agentId', '?')))
print(f" - {label}: {name}")
if len(items) > 5:
print(f" ... and {len(items) - 5} more")
Step 6: Query the hierarchy paths
Visualize the full hierarchy for a sample Thing using Cypher.
[ ]:
# Get first Thing
result = graph.query(
"MATCH (t:Thing) RETURN t.identifier AS id ORDER BY id LIMIT 1"
)
if result:
thing_id = result[0]['id']
sub_section(f"Hierarchy: Thing({thing_id})")
# Query full hierarchy
result = graph.query(
f"MATCH path = (t:Thing {{identifier: '{thing_id}'}})-[*1..6]->(n) "
"RETURN path"
)
if result:
for row in result:
path = row.get("path")
if path:
if isinstance(path, dict):
nodes = path.get("nodes", [])
rels = path.get("relationships", [])
else:
nodes = path.nodes
rels = path.relationships
parts = []
for node in nodes:
if isinstance(node, dict):
label = list(node.get("labels", ["Node"]))[0]
props = node.get("properties", {})
else:
label = list(node.labels)[0] if hasattr(node, 'labels') else 'Node'
props = dict(node)
name = props.get("fullName", props.get("name", props.get("agentId", props.get("recordResourceId", str(getattr(node, 'element_id', '?'))))))
parts.append(f"{label}({name})")
for rel in rels:
if isinstance(rel, dict):
parts.append(f"-[{rel.get('type', '')}]->")
else:
parts.append(f"-[{rel.type}]->")
print(f" {' → '.join(parts)}")
Step 7: Inspect specific entities
Look at the properties of loaded entities to understand the data model.
[ ]:
# Show sample Agent entities
sub_section("Sample Agent entities")
result = graph.query(
"MATCH (a:Agent) RETURN a.agentId AS id, a.label AS label, "
"a.beginningDate AS birth, a.endDate AS death "
"ORDER BY id LIMIT 5"
)
for row in result:
print(f" {row['id']}: {row['label']}")
print(f" Born: {row['birth'] or 'N/A'}, Died: {row['death'] or 'N/A'}")
# Show sample RecordResource entities
sub_section("Sample RecordResource entities")
result = graph.query(
"MATCH (r:RecordResource) RETURN r.recordResourceId AS id, "
"r.title AS title, r.beginningDate AS from_date, r.endDate AS to_date "
"ORDER BY id LIMIT 5"
)
for row in result:
print(f" {row['id']}: {row['title'] or 'No title'}")
print(f" Date range: {row['from_date'] or 'N/A'} — {row['to_date'] or 'N/A'}")
Step 8: Summary statistics
Show the final state of the RiC-O graph.
[ ]:
section("Final summary")
# Nodes by label
result = graph.query(
"MATCH (n) RETURN labels(n)[0] AS label, count(n) AS cnt "
"ORDER BY cnt DESC"
)
print("Nodes by label:")
for row in result:
print(f" {row['label']}: {row['cnt']}")
# Propagation edges
result = graph.query(
"MATCH ()-[r]->() WHERE r._propagate = TRUE "
"RETURN type(r) AS rel_type, count(r) AS cnt "
"ORDER BY cnt DESC"
)
print("\nPropagation edges by type:")
for row in result:
print(f" {row['rel_type']}: {row['cnt']}")
# All Things
result = graph.query(
"MATCH (t:Thing) RETURN t.identifier AS id, t.name AS name "
"ORDER BY id"
)
print("\nAll Things:")
for row in result:
print(f" {row['id']}: {row['name']}")
# Subdocument counts
print("\nSubdocuments per Thing:")
for row in graph.query("MATCH (t:Thing) RETURN t.identifier AS id ORDER BY id"):
thing_node = Thing(pk={"identifier": row['id']})
subdocs = graph.get_subdocuments(thing_node)
print(f" {row['id']}: {len(subdocs)} subdocuments")
print("\n" + "=" * 60)
print(" ✅ RiC-O demo completed successfully!")
print("=" * 60)
Cleanup
Close the Neo4j connection when done.
[ ]:
graph.close()
print("✅ Neo4j connection closed.")