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

import importlib.util

# Check if cvcdocdb is installed
package_to_check = 'cvcdocdb'
spec = importlib.util.find_spec(package_to_check)

if spec is None:
    print(f'⚠️ {package_to_check} not installed. Installing...')
    %pip install -q --upgrade cvcdocdb-tools
    print("✅ Installation complete.")
else:
    print(f'✅ {package_to_check} already installed. Skipping.')

✅ cvcdocdb already installed. Skipping.

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 cvcdocdb package to path for development
_repo_root = Path.cwd()
if not (_repo_root / "cvcdocdb").exists():
    for parent in Path.cwd().parents:
        if (parent / "cvcdocdb").exists():
            _repo_root = parent
            break
_cvcdocdb_path = str(_repo_root / "cvcdocdb")
if _cvcdocdb_path not in sys.path:
    sys.path.insert(0, _cvcdocdb_path)
    print(f"✅ Added {_cvcdocdb_path} to sys.path")

import json

from cvcdocdb import Neo4jGraph, NetworkXGraph
from cvcdocdb.exemples import load_got_characters
from cvcdocdb.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/cvcdocdb-tools/.env
   Found 5 variables
✅ Added /Users/oriol/Desenvolupament/cvcdocdb-tools/cvcdocdb to sys.path
✅ All imports successful.

Step 1: Connect to Neo4j

Connect to the DEV Neo4j instance and verify the connection.

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: 57

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.

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
  Valor: 5
  IndividuPadro: 3
  LlocPadro: 2
  NodeA: 1
  NodeB: 1
  TestNode: 1
  TransientNode: 1
  TEST: 1
  Document: 1
  Page: 1

--- Relations by type ---
  MEMBER_OF: 25
  COGNOM1: 3
  NOM: 3
  CONNECTS_TO: 1
  HAS_PAGE: 1

Step 3: Generate YAML schema

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

output_dir = repo_path("cvcdocdb/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/cvcdocdb-tools/cvcdocdb/exemples/generated/got_schema.yaml
Size: 3845 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
    Document:
      class_name: Document
      base_class: Node
      properties:
        _weak_init_done: boolean
        doc_id: 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.

# 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/cvcdocdb-tools/cvcdocdb/exemples/generated/got_entities.py
Size: 11259 bytes

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

  from __future__ import annotations

  from typing import Any, Dict, List, Optional

  from cvcdocdb.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 Document(Node):
  ...

Step 5: Initialize propagation properties

Scan the entire Neo4j graph and initialize propagation properties:

• is_weak: marks nodes that are children of _propagate edges

• _propagate: marks nodes that have cascade-delete enabled

• parent_relation: stores the edge type linking parent → child

• _weak_init_done: marks strong nodes whose weak children are initialized

This uses init_propagation() which:

1. Finds all edges with _propagate = TRUE

2. Marks the child nodes as is_weak = TRUE

3. Skips nodes already marked via _weak_init_done

# 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:          57
  Nodes with is_weak:   1
  Nodes with _weak_init_done: 17

Running init_propagation()...
Result: True

--- After init_propagation() ---
  Total nodes:          57
  Nodes with is_weak:   1
  Nodes with _weak_init_done: 57
  Edges with _propagate: 1

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.

from cvcdocdb.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

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
Cell In[8], line 54
     50 print(f"  Weak nodes:   Section(intro), Section(characters)")
     51 print(f"  Weak nodes:   Page(1) → intro, Page(1) → characters")
     52
     53 # Create atomically
---> 54 doc_id = graph.create_group(
     55     strong_node=doc,
     56     weak_nodes=[section1, section2, page1, page2],
     57 )

File ~/opt/anaconda3/envs/cvcdocdb-env/lib/python3.11/site-packages/cvcdocdb/neo4j_graph.py:700, in Neo4jGraph.create_group(self, strong_node, weak_nodes, weak_relations, **kwargs)
    697     return result
    699 try:
--> 700     return session.write_transaction(_create_group_tx)
    701 except ConstraintError as err:
    702     raise RuntimeError("Duplicate key: " + err.message) from err

AttributeError: 'Session' object has no attribute 'write_transaction'

Step 7: Verify the group

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

# 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['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['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['content'] or '')[:50]
    print(f"  Page: '{content}...' is_weak={row.get('is_weak')}, _propagate={row.get('_propagate')}")

Step 8: Sample queries

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

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

# 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 type(a) AS from_label, type(r) AS rel_type, "
    "type(b) 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']}")

# 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:
            nodes = path.nodes
            rels = path.relationships
            parts = []
            for i, node in enumerate(nodes):
                label = list(node.labels)[0] if hasattr(node, 'labels') else 'Node'
                props = dict(node)
                name = props.get('title', props.get('name', str(node.element_id)))
                parts.append(f"{label}({name})")
            for rel in rels:
                parts.append(f"-[{rel.type}]->")
            print(f"  {' → '.join(parts)}")

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

Cleanup

Close the Neo4j connection when done.

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