Propagation Properties — Full Neo4j Demo

This notebook demonstrates the complete workflow for working with propagation properties (_propagate, is_weak, parent_relation, _weak_init_done) on a real Neo4j database.

Steps:

  1. Connect to Neo4j DEV and load the Game of Thrones dataset

  2. Generate YAML schema + Python entity classes

  3. Initialize propagation properties on existing data

  4. Demonstrate create_group() — atomic group creation

  5. Run sample queries to verify everything works

Prerequisites

  • A running Neo4j instance (DEV target)

  • Environment variables: NEO4J_DEV_URL, NEO4J_DEV_USER, NEO4J_DEV_PASSWORD

  • Default connection: bolt://localhost:7687 / neo4j / secret

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")

import json

from drm import Neo4jGraph, NetworkXGraph
from drm.exemples import load_got_characters
from drm.schema_gen import generate_classes


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} ---")


# Resolve paths relative to repo root
def repo_path(rel):
    """Return a Path relative to the repository root."""
    return _repo_root / rel

print("✅ All imports successful.")

✅ Loaded .env from /Users/oriol/Desenvolupament/drm-tools/.env
   Found 5 variables
✅ Added /Users/oriol/Desenvolupament/drm-tools to sys.path
✅ 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: 45

Step 1.5: Clear existing data

Delete all nodes and relationships before loading fresh GOT 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 Game of Thrones dataset

Load character and house data from thronesapi.com. Falls back to bundled data if the API is unavailable.

[5]:

stats = load_got_characters(graph, limit=25)
print("Loaded:", json.dumps(stats, indent=2))

# 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']}")

Loaded: {
  "characters": 25,
  "houses": 15,
  "member_of": 25
}

--- Nodes by label ---
  Character: 25
  House: 15

--- Relations by type ---
  MEMBER_OF: 25

Step 3: Generate YAML schema

Introspect the Neo4j database and generate a YAML schema description. This captures labels, properties, relationships, and counts.

[6]:

output_dir = repo_path("drm/exemples/generated")
output_dir.mkdir(exist_ok=True)

yaml_content = graph.schema_yaml(db_name="got")
yaml_path = output_dir / "got_schema.yaml"
yaml_path.write_text(yaml_content)

print(f"Written: {yaml_path}")
print(f"Size: {len(yaml_content)} bytes")

print("\nSchema preview:")
for line in yaml_content.split("\n")[:20]:
    print(f"  {line}")
print("  ...")

Written: /Users/oriol/Desenvolupament/drm-tools/drm/exemples/generated/got_schema.yaml
Size: 820 bytes

Schema preview:
  # Schema generated from: got
  # Generated at: 2026-07-12

  labels:
    Character:
      class_name: Character
      base_class: Node
      properties:
        character_id: integer
        name: string
        source: string
        title: string
      primary_key: ['name', 'character_id']
      count: 25
    House:
      class_name: House
      base_class: Node
      properties:
        house_name: string
        name: string
  ...

Step 4: Generate Python entity classes

Generate Python classes from the YAML schema. This creates Node, WeakNode, Relation, and WeakRelation subclasses for each label and relationship type in the database.

[7]:

# Read the YAML file content (generate_classes expects YAML string, not file path)
yaml_content = yaml_path.read_text()
code = generate_classes(yaml_content)

py_path = output_dir / "got_entities.py"
py_path.write_text(code)

print(f"Written: {py_path}")
print(f"Size: {len(code)} bytes")

print("\nGenerated code preview:")
for line in code.split("\n")[:25]:
    print(f"  {line}")
print("  ...")

Written: /Users/oriol/Desenvolupament/drm-tools/drm/exemples/generated/got_entities.py
Size: 2520 bytes

Generated code preview:
  """Auto-generated entity classes from schema."""

  from __future__ import annotations

  from typing import Any, Dict, List, Optional

  from drm.base import Node, WeakNode, Relation, WeakRelation


  class Character(Node):
      """Auto-generated entity class.."""

      def __init__(self, pk: Dict[str, Any], **kwargs: Any) -> None:
          """Initialize a Character node.

          Args:
              pk: Primary key dict with fields: name, character_id.
              **kwargs: Additional properties and attributes.
          """
          super().__init__(pk=pk, main_label="Character", **kwargs)
          self.character_id = kwargs.get("character_id", 'integer')
          self.name = kwargs.get("name", 'string')
          self.source = kwargs.get("source", 'string')
          self.title = kwargs.get("title", 'string')
  class House(Node):
  ...

Step 5: Initialize propagation properties

Scan the entire Neo4j graph and initialize propagation properties:

  • is_weak: marks nodes that are children of edges with _propagate = TRUE

  • _propagate: marks edges that have cascade-delete enabled

  • parent_relation: stores the edge type linking parent → child

  • _weak_init_done: marks nodes whose propagation properties have been initialized

This uses init_propagation() which:

  1. Finds all edges with _propagate = TRUE

  2. Marks the child nodes as is_weak = TRUE (only for edges that have _propagate)

  3. Skips nodes already initialized via _weak_init_done

Note: is_weak is only set on nodes that are direct children of edges marked with _propagate = TRUE. Nodes not connected via such edges will remain unmarked.

[8]:

# 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}")


--- Before init_propagation() ---
  Total nodes:          40
  Nodes with is_weak:   0
  Nodes with _weak_init_done: 0

Running init_propagation()...
Result: True

--- After init_propagation() ---
  Total nodes:          40
  Nodes with is_weak:   0
  Nodes with _weak_init_done: 40
  Edges with _propagate: 0

Step 6: Transactional group creation

Use create_group() to atomically create a strong node with its WeakNodes and WeakRelations. The entire group is created in a single Neo4j transaction — if any step fails, everything is rolled back.

We create a Document → Section → Page hierarchy.

[9]:

from drm.base import Node, WeakNode

# Create the strong node (Document)
doc = Node(
    pk={"doc_id": "GOT-DEMO-001"},
    main_label="Document",
    title="Game of Thrones — Episode Guide",
    author="Demo Script",
)

# Create weak nodes (Sections and Pages)
section1 = WeakNode(
    parent=doc,
    parent_relation="HAS_SECTION",
    pk={"section": "intro"},
    main_label="Section",
    title="Introduction",
    order=1,
)

section2 = WeakNode(
    parent=doc,
    parent_relation="HAS_SECTION",
    pk={"section": "characters"},
    main_label="Section",
    title="Main Characters",
    order=2,
)

page1 = WeakNode(
    parent=section1,
    parent_relation="HAS_PAGE",
    pk={"page": 1},
    main_label="Page",
    content="Welcome to the Game of Thrones episode guide.",
    order=1,
)

page2 = WeakNode(
    parent=section2,
    parent_relation="HAS_PAGE",
    pk={"page": 1},
    main_label="Page",
    content="Jon Snow, Daenerys Targaryen, Tyrion Lannister...",
    order=1,
)

print("Creating group:")
print(f"  Strong node:  Document(GOT-DEMO-001)")
print(f"  Weak nodes:   Section(intro), Section(characters)")
print(f"  Weak nodes:   Page(1) → intro, Page(1) → characters")

# Create atomically
doc_id = graph.create_group(
    strong_node=doc,
    weak_nodes=[section1, section2, page1, page2],
)

print(f"\n✅ Group created (strong node id: {doc_id})")

Creating group:
  Strong node:  Document(GOT-DEMO-001)
  Weak nodes:   Section(intro), Section(characters)
  Weak nodes:   Page(1) → intro, Page(1) → characters

✅ Group created (strong node id: 833)

Step 7: Verify the group

Query the database to verify the group was created correctly and check the propagation properties.

[10]:

# Verify Document node
sub_section("Document node")
result = graph.query(
    "MATCH (n) WHERE n.doc_id = 'GOT-DEMO-001' "
    "RETURN labels(n) AS labels, n.title, n._weak_init_done"
)
for row in result:
    print(f"  Labels: {row['labels']}")
    print(f"  Title:  {row.get('title')}")
    print(f"  _weak_init_done: {row.get('_weak_init_done')}")

# Verify Section nodes
sub_section("Section nodes")
result = graph.query(
    "MATCH (n:Section) WHERE n._weak_init_done IS NULL "
    "RETURN labels(n) AS labels, n.title, n.is_weak, n._propagate"
)
for row in result:
    print(f"  {row.get('title')}: is_weak={row.get('is_weak')}, _propagate={row.get('_propagate')}")

# Verify Page nodes
sub_section("Page nodes")
result = graph.query(
    "MATCH (n:Page) WHERE n._weak_init_done IS NULL "
    "RETURN labels(n) AS labels, n.content, n.is_weak, n._propagate"
)
for row in result:
    content = (row.get('content') or '')[:50]
    print(f"  Page: '{content}...' is_weak={row.get('is_weak')}, _propagate={row.get('_propagate')}")


--- Document node ---
  Labels: ['Document']
  Title:  None
  _weak_init_done: None
  Labels: ['Section']
  Title:  None
  _weak_init_done: None
  Labels: ['Section']
  Title:  None
  _weak_init_done: None
  Labels: ['Page']
  Title:  None
  _weak_init_done: None
  Labels: ['Page']
  Title:  None
  _weak_init_done: None

--- Section nodes ---
  None: is_weak=None, _propagate=None
  None: is_weak=None, _propagate=None

--- Page nodes ---
  Page: '...' is_weak=None, _propagate=None
  Page: '...' is_weak=None, _propagate=None

Step 8: Sample queries

Run several queries to demonstrate the initialized graph and propagation properties.

[11]:

# Query 1: Characters by House
sub_section("Characters by House")
result = graph.query(
    "MATCH (c:Character)-[:MEMBER_OF]->(h:House) "
    "RETURN h.name AS house, collect(c.name) AS characters "
    "ORDER BY house"
)
for row in result:
    chars = row['characters'][:5]
    print(f"  {row['house']}: {', '.join(chars)}")
    if len(row['characters']) > 5:
        print(f"    ... and {len(row['characters']) - 5} more")


--- Characters by House ---
  Free Folk: Ygritte
  House Baelish: Petyr Baelish
  House Baratheon: Stannis Baratheon, Robert Baratheon
  House Clegane: The Hound
  House Greyjoy: Theon Greyjoy
  House Lanister: Tyrion Lannister, Joffrey Baratheon
  House Lannister: Cersei Lannister, Jamie Lannister
  House Seaworth: Davos Seaworth
  House Stark: Rob Stark, Catelyn Stark, Ned Stark, Brandon Stark, Sansa Stark
    ... and 2 more
  House Targaryen: Khal Drogo, Daenerys Targaryen
  House Tarly: Samwell Tarly
  House Tyrell: Margaery Tyrell
  Naathi: Missandei
  Tarth: Brienne of Tarth
  Unknown: Varys

[12]:

# Query 2: Nodes with propagation properties
sub_section("Nodes with propagation properties")
result = graph.query(
    "MATCH (n) WHERE n.is_weak = TRUE "
    "RETURN labels(n) AS label, count(n) AS count "
    "ORDER BY count DESC"
)
for row in result:
    print(f"  {row['label']}: {row['count']} nodes")

# Query 3: Edges with _propagate
sub_section("Edges with _propagate")
result = graph.query(
    "MATCH (a)-[r]->(b) WHERE r._propagate = TRUE "
    "RETURN labels(a)[0] AS from_label, type(r) AS rel_type, "
    "labels(b)[0] AS to_label, count(*) AS count "
    "ORDER BY count DESC"
)
for row in result:
    print(f"  {row['from_label']} -[{row['rel_type']}]-> {row['to_label']}: {row['count']}")


--- Nodes with propagation properties ---

--- Edges with _propagate ---
  Document -[HAS_SECTION]-> Section: 2
  Document -[HAS_PAGE]-> Page: 2

[13]:

# Query 4: Document group hierarchy
sub_section("Document group hierarchy")
result = graph.query(
    "MATCH path = (d:Document {doc_id: 'GOT-DEMO-001'})-[*1..3]->(n) "
    "RETURN path"
)
if result:
    for row in result:
        path = row.get("path")
        if path:
            # Handle both native objects (driver 4.x) and dicts (driver 6.x)
            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]
                    name = node.get("title", node.get("name", str(node)))
                else:
                    label = list(node.labels)[0] if hasattr(node, 'labels') else 'Node'
                    name = node.get('title', node.get('name', str(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)}")


--- Document group hierarchy ---
  Document({'labels': ['Document'], 'properties': {'_weak_init_done': True, 'author': 'Demo Script', 'title': 'Game of Thrones — Episode Guide', 'doc_id': 'GOT-DEMO-001'}}) → Section({'labels': ['Section'], 'properties': {'section': 'intro', 'title': 'Introduction', 'doc_id': 'GOT-DEMO-001', 'order': 1}}) → -[HAS_SECTION]->
  Document({'labels': ['Document'], 'properties': {'_weak_init_done': True, 'author': 'Demo Script', 'title': 'Game of Thrones — Episode Guide', 'doc_id': 'GOT-DEMO-001'}}) → Page({'labels': ['Page'], 'properties': {'section': 'characters', 'page': 1, 'doc_id': 'GOT-DEMO-001', 'content': 'Jon Snow, Daenerys Targaryen, Tyrion Lannister...', 'order': 1}}) → -[HAS_PAGE]->
  Document({'labels': ['Document'], 'properties': {'_weak_init_done': True, 'author': 'Demo Script', 'title': 'Game of Thrones — Episode Guide', 'doc_id': 'GOT-DEMO-001'}}) → Page({'labels': ['Page'], 'properties': {'section': 'intro', 'page': 1, 'doc_id': 'GOT-DEMO-001', 'content': 'Welcome to the Game of Thrones episode guide.', 'order': 1}}) → -[HAS_PAGE]->
  Document({'labels': ['Document'], 'properties': {'_weak_init_done': True, 'author': 'Demo Script', 'title': 'Game of Thrones — Episode Guide', 'doc_id': 'GOT-DEMO-001'}}) → Section({'labels': ['Section'], 'properties': {'section': 'characters', 'title': 'Main Characters', 'doc_id': 'GOT-DEMO-001', 'order': 2}}) → -[HAS_SECTION]->

[14]:

# Query 5: Summary statistics
sub_section("Summary statistics")

result = graph.query(
    "MATCH (n) "
    "RETURN count(n) AS total_nodes, "
    "count(n.is_weak) AS weak_nodes, "
    "count(n._weak_init_done) AS initialized_nodes"
)
row = result[0] if result else {}
print(f"  Total nodes:            {row.get('total_nodes', 0)}")
print(f"  Weak nodes:             {row.get('weak_nodes', 0)}")
print(f"  Initialized nodes:      {row.get('initialized_nodes', 0)}")

result = graph.query(
    "MATCH ()-[r]->() "
    "RETURN count(r) AS total_edges, "
    "count(r._propagate) AS propagate_edges"
)
row = result[0] if result else {}
print(f"  Total edges:            {row.get('total_edges', 0)}")
print(f"  Propagate edges:        {row.get('propagate_edges', 0)}")

print("\n" + "=" * 60)
print("  ✅ Demo completed successfully!")
print("=" * 60)


--- Summary statistics ---
  Total nodes:            45
  Weak nodes:             0
  Initialized nodes:      41
  Total edges:            29
  Propagate edges:        4

============================================================
  ✅ Demo completed successfully!
============================================================

Step 9: Get subdocuments

Given a strong document node, retrieve all its subdocuments (sections, pages, etc.) connected through _propagate edges.

[15]:

# get_subdocuments: find all subdocuments of a strong node
sub_section("9a: get_subdocuments for GOT-DEMO-001")

strong_doc = Node(pk={"doc_id": "GOT-DEMO-001"}, main_label="Document")
subdocuments = graph.get_subdocuments(strong_doc)

print(f"  Found {len(subdocuments)} subdocuments:")
for sd in subdocuments:
    name = sd["pk"].get("title", sd["pk"].get("section", sd["pk"].get("page", "?")))
    print(f"    - {sd['label']}: {name}")


--- 9a: get_subdocuments for GOT-DEMO-001 ---
  Found 4 subdocuments:
    - Section: Introduction
    - Page: characters
    - Page: intro
    - Section: Main Characters

[16]:

# Final summary
sub_section("Final summary")

result = graph.query(
    "MATCH (n) "
    "RETURN count(n) AS total_nodes, "
    "count(n.is_weak) AS weak_nodes, "
    "count(n._weak_init_done) AS initialized_nodes"
)
row = result[0] if result else {}
print(f"  Total nodes:            {row.get('total_nodes', 0)}")
print(f"  Weak nodes:             {row.get('weak_nodes', 0)}")
print(f"  Initialized nodes:      {row.get('initialized_nodes', 0)}")

result = graph.query(
    "MATCH ()-[r]->() "
    "RETURN count(r) AS total_edges, "
    "count(r._propagate) AS propagate_edges"
)
row = result[0] if result else {}
print(f"  Total edges:            {row.get('total_edges', 0)}")
print(f"  Propagate edges:        {row.get('propagate_edges', 0)}")

print("\n" + "=" * 60)
print("  Demo completed successfully!")
print("=" * 60)


--- Final summary ---
  Total nodes:            45
  Weak nodes:             0
  Initialized nodes:      41
  Total edges:            29
  Propagate edges:        4

============================================================
  Demo completed successfully!
============================================================

Cleanup

Close the Neo4j connection when done.

[17]:

graph.close()
print("✅ Neo4j connection closed.")

✅ Neo4j connection closed.