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

> Learn how the Python SDK's GraphLoader ingests code into Memgraph for code-graph-rag. Discover efficient batch processing and bulk Cypher queries for seamless data ingestion.

- Repository: [Vitali Avagyan/code-graph-rag](https://github.com/vitali87/code-graph-rag)
- Tags: how-to-guide
- Published: 2026-08-16

---

**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`](https://github.com/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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/services/graph_service.py) | Manages Bolt connections, buffers data, and executes bulk Cypher statements |
| **GraphUpdater** | [`codebase_rag/graph_updater.py`](https://github.com/vitali87/code-graph-rag/blob/main/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

```python
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`](https://github.com/vitali87/code-graph-rag/blob/main/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

```python
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:

```python
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`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/graph_updater.py), it connects parsers, the loader, and the ingestor.

### Core Orchestration Flow

```python
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

```python
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`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/graph_loader.py) | JSON parsing and in-memory graph representation |
| [`codebase_rag/services/graph_service.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/services/graph_service.py) | `MemgraphIngestor` implementation with batching and Cypher generation |
| [`codebase_rag/graph_updater.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/graph_updater.py) | End-to-end orchestration via `GraphUpdater` |
| [`codebase_rag/constants.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/constants.py) | Node labels, relationship types, and schema constants |
| [`codebase_rag/logs.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/logs.py) | Structured logging for ingestion operations |
| [`codebase_rag/exceptions.py`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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.