How the Python SDK's GraphLoader Ingests Code into Memgraph for code-graph-rag

The Python SDK's GraphLoader ingests code into Memgraph by reading a JSON export of repository structure, buffering nodes and relationships in batches, and executing bulk Cypher MERGE queries via the MemgraphIngestor class.

The code-graph-rag project transforms static code repositories into queryable knowledge graphs. This article explains the complete ingestion pipeline—from exported JSON to live Memgraph database—using the actual source implementation from vitali87/code-graph-rag.

Three Core Components of the Ingestion Pipeline

The Python SDK splits responsibilities across three main classes:

Component	Location	Primary Role
GraphLoader	codebase_rag/graph_loader.py	Parses JSON exports and provides in-memory node/relationship objects with fast lookups
MemgraphIngestor	codebase_rag/services/graph_service.py	Manages Bolt connections, buffers data, and executes bulk Cypher statements
GraphUpdater	codebase_rag/graph_updater.py	Orchestrates the full pipeline from parsing to database insertion

Loading Exported Graph Data with GraphLoader

The GraphLoader class reads repository structure from JSON files produced by tree-sitter parsers. It converts raw JSON into typed GraphNode and GraphRelationship objects.

Key Methods in graph_loader.py

• load() — reads the JSON file and populates self.nodes and self.relationships
• find_nodes_by_label(label) — filters nodes by label for targeted processing
• find_node_by_property(key, value) — retrieves specific nodes by unique properties
• summary() — returns ingestion statistics including label distributions

from codebase_rag.graph_loader import GraphLoader

loader = GraphLoader("/tmp/repo_graph.json")
loader.load()

# Access loaded data

print(f"Nodes: {len(loader.nodes)}")
print(f"Relationships: {len(loader.relationships)}")
print(loader.summary())

The loader maintains indexed lookups, enabling efficient access patterns when streaming large graphs to the database.

Batching and Bulk Insertion with MemgraphIngestor

The MemgraphIngestor class in codebase_rag/services/graph_service.py handles the actual database communication. It implements the IngestorProtocol interface and optimizes for throughput through batching and parallel execution.

Connection and Buffer Management

from codebase_rag.services.graph_service import MemgraphIngestor

with MemgraphIngestor(
    host="localhost",
    port=7687,
    batch_size=500,
    username="",
    password=""
) as ingestor:
    # Nodes and relationships are buffered automatically

    pass  # Flushes on context exit

The constructor validates credentials, initializes a thread pool for parallel flushing, and prepares empty buffers for nodes and relationships.

Buffering Methods

Method	Signature	Purpose
ensure_node_batch()	ensure_node_batch(label, properties)	Appends (label, properties) to node_buffer
ensure_relationship_batch()	ensure_relationship_batch(from_ref, rel_type, to_ref, properties)	Appends relationship pattern to relationship_buffer
flush_nodes()	flush_nodes()	Builds and executes bulk MERGE/CREATE for buffered nodes
flush_relationships()	flush_relationships()	Builds and executes bulk MERGE/CREATE for buffered relationships
flush_all()	flush_all()	Forces complete flush of all buffers and closes connection cleanly

Relationship Pattern Keys

The ensure_relationship_batch() method accepts typed references:

ingestor.ensure_relationship_batch(
    (from_label, merge_key, merge_value),  # Source node pattern

    relationship_type,
    (to_label, merge_key, merge_value),    # Target node pattern

    properties_dict
)

This pattern grouping enables efficient bulk MERGE operations that avoid creating duplicate edges.

Orchestrating the Pipeline with GraphUpdater

The GraphUpdater class drives the complete ingestion flow without manual buffer management. Located in codebase_rag/graph_updater.py, it connects parsers, the loader, and the ingestor.

Core Orchestration Flow

from codebase_rag.graph_updater import GraphUpdater
from codebase_rag.services.graph_service import MemgraphIngestor
from codebase_rag.parser_loader import load_parsers_and_queries
from pathlib import Path

repo_path = Path("/path/to/my/repo")
parsers, queries = load_parsers_and_queries()

with MemgraphIngestor(host="localhost", port=7687) as ingestor:
    updater = GraphUpdater(
        ingestor=ingestor,      # Any IngestorProtocol implementation

        repo_path=repo_path,
        parsers=parsers,
        queries=queries,
    )
    updater.run(force=False)    # Full parse → graph → ingest pipeline

The run() method internally:

1. Parses source files using tree-sitter via parsers
2. Builds the JSON graph representation
3. Streams nodes through _sink.ensure_node_batch()
4. Streams relationships through _sink.ensure_relationship_batch()
5. Calls flush_all() on completion

Step-by-Step Ingestion Flow

The complete GraphLoader → Memgraph pipeline follows this sequence:

1. Export — Repository parsing produces a JSON file with nodes, relationships, and metadata
2. Load — GraphLoader(file_path) deserializes JSON into GraphNode and GraphRelationship objects
3. Create Ingestor — MemgraphIngestor establishes Bolt connection and initializes thread pool
4. Buffer Nodes — ensure_node_batch(label, properties) appends to node_buffer until batch_size
5. Buffer Relationships — ensure_relationship_batch() groups edges by pattern for efficient merging
6. Flush in Bulk — Buffer thresholds trigger flush_nodes() and flush_relationships(), which build Cypher queries using UNWIND for batch execution
7. Commit — flush_all() ensures remaining data persists and closes the connection

Performance Optimizations

• Batch sizing — Configurable batch_size parameter (default typically 500-1000) minimizes round-trips
• Parallel flushing — Thread pool enables concurrent Cypher execution
• Memory limits — _apply_memory_limit() helper adds timeout clauses to prevent runaway queries

Complete Ingestion Example

from codebase_rag.graph_loader import GraphLoader
from codebase_rag.services.graph_service import MemgraphIngestor

# 1. Load exported graph

loader = GraphLoader("/tmp/repo_graph.json")
loader.load()

# 2. Ingest to Memgraph with explicit batch control

with MemgraphIngestor(
    host="localhost",
    port=7687,
    batch_size=500
) as ingestor:
    
    # Stream all nodes

    for node in loader.nodes:
        ingestor.ensure_node_batch(
            node.labels[0],
            node.properties
        )
    
    # Stream all relationships with merge keys

    for rel in loader.relationships:
        ingestor.ensure_relationship_batch(
            (rel.from_label, "qualified_name", rel.from_id),
            rel.type,
            (rel.to_label, "qualified_name", rel.to_id),
            rel.properties,
        )
    
    # Final flush (automatic on exit, but explicit for clarity)

    ingestor.flush_all()

Key Source Files Reference

File	Purpose
codebase_rag/graph_loader.py	JSON parsing and in-memory graph representation
codebase_rag/services/graph_service.py	MemgraphIngestor implementation with batching and Cypher generation
codebase_rag/graph_updater.py	End-to-end orchestration via GraphUpdater
codebase_rag/constants.py	Node labels, relationship types, and schema constants
codebase_rag/logs.py	Structured logging for ingestion operations
codebase_rag/exceptions.py	Custom error types for pipeline failures

Summary

• GraphLoader transforms JSON exports into queryable GraphNode and GraphRelationship objects
• MemgraphIngestor implements efficient bulk insertion through buffering, batching, and parallel Cypher execution
• GraphUpdater provides a higher-level API that automates the entire parse-to-database flow
• The SDK uses pattern-based relationship merging with qualified_name keys to prevent duplicate edges
• Thread pool parallelism and configurable batch sizes enable scaling to hundreds of thousands of nodes and edges

Frequently Asked Questions

How does GraphLoader handle large JSON files?

GraphLoader loads the entire JSON into memory using standard json.load(), then constructs indexed lists of nodes and relationships. For extremely large repositories, use the GraphUpdater streaming API instead, which processes files incrementally without holding the full graph in memory.

What Cypher operations does MemgraphIngestor use for insertion?

The ingestor generates bulk MERGE queries by default (configurable to CREATE), using UNWIND to batch multiple nodes or relationships per transaction. This reduces network overhead compared to individual CREATE statements.

Can I use a different database backend with the same GraphLoader?

Yes. GraphLoader is database-agnostic—it only produces Python objects. The IngestorProtocol abstract base class in graph_service.py allows swapping MemgraphIngestor for other implementations (Neo4j, Amazon Neptune, etc.) without changing loader or updater code.

How are relationship duplicates prevented during ingestion?

MemgraphIngestor groups relationships by (from_label, merge_key, merge_value, rel_type, to_label, to_key, to_value) patterns. The generated Cypher uses MERGE on these specific property combinations, ensuring identical source-target-type triples collapse to single edges while distinct edges with different properties persist.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}