Code Property Graph

Code Property Graphs — Joern/Semgrep extraction, AI vulnerability enrichment, risk scoring.

This package depends on AWS GraphRAG Toolkit (graphrag-toolkit-lexical-graph) for graph storage, vector indexing, and retrieval.

Installation

pip install graphrag-codeproperty-graph

Dependencies

• graphrag-document-graph>=3.1.1 — Document graph infrastructure (typed nodes, Cypher builders, multi-tenancy)
• boto3>=1.28.0 — AWS SDK for S3 manifest storage and service integrations

Dependency Chain

graphrag-codeproperty-graph
└── graphrag-document-graph>=3.0.8
    └── graphrag-toolkit-lexical-graph>=3.18.0  (AWS foundation)
        ├── Neptune graph storage
        ├── OpenSearch Serverless vector indexing
        └── Entity resolution & retrieval

Quick Start

CPGService — Delta Node-Level Filtering (Recommended)

from codeproperty_graph import CPGService

# Only writes changed/added method subtrees to Neptune
service = CPGService(bucket="graphrag-artifacts-705909755305")

result = await service.ingest(
    repo="my-service",
    job_id="build-456",
    tenant_id="tenant_abc123",
    nodes_data=joern_nodes,
    edges_data=joern_edges,
    nodes_path="s3://bucket/cpg-exports/my-service/build-456/nodes.json",
    edges_path="s3://bucket/cpg-exports/my-service/build-456/edges.json",
    graph_store=neptune_store,
    write_fn=batch_write_function,
    remove_fn=remove_method_subgraph,  # optional: surgical deletion
)

# result: {"status": "SKIPPED"} — no changes
# result: {"status": "INGESTED", "delta": "+5 -2 ~3 =150",
#           "filtered_nodes": 47, "total_nodes": 5123}

DeltaIngestor — Full Artifact Pipeline

from codeproperty_graph import DeltaIngestor

# Full pipeline: graph + vectors + summaries + code slices
ingestor = DeltaIngestor(bucket="graphrag-artifacts-705909755305")

result = await ingestor.ingest(
    repo="my-service",
    job_id="build-456",
    tenant_id="tenant_abc123",
    nodes_data=joern_nodes,
    edges_data=joern_edges,
    nodes_path="s3://bucket/cpg-exports/my-service/build-456/nodes.json",
    edges_path="s3://bucket/cpg-exports/my-service/build-456/edges.json",
    graph_store=neptune_store,
    write_fn=batch_write_function,
)

# result: {"status": "SKIPPED"} or {"status": "INGESTED", "delta": "+5 -2 ~3 =150"}

Graph Diff — Compare CPG States

from codeproperty_graph import GraphDiff, CPGNode

# Compare current vs previous code analysis
diff = GraphDiff.compare(
    current_nodes=current_cpg_nodes,
    previous_nodes=previous_cpg_nodes,
)

print(f"Added: {len(diff.added)}")
print(f"Removed: {len(diff.removed)}")
print(f"Modified: {len(diff.modified)}")
print(f"Unchanged: {len(diff.unchanged)}")

Manifest Management

from codeproperty_graph import ManifestManager

# Track CPG state per repository in S3
manager = ManifestManager(bucket="graphrag-artifacts-705909755305")

# Save manifest after successful ingestion
await manager.save(repo="my-service", job_id="build-456", signatures=method_signatures)

# Load previous manifest for diff comparison
previous = await manager.load(repo="my-service")

Package Structure

src/codeproperty_graph/
├── __init__.py           # Public API (30 exports)
├── models.py             # CPGNode, CPGEdge, Manifest, VectorRecord, SummaryRecord, CodeSliceRecord
├── schema.py             # Full Joern CPG schema (20 node types, 14 edge types)
├── graph_diff.py         # Compare two CPG states by method signature
├── manifest_manager.py   # S3-backed state tracking with optimistic locking
├── delta_ingestor.py     # Full artifact pipeline orchestration
├── cpg_service.py        # Node-level delta filtering (only write changed methods)
├── tenant_ops.py         # Lifecycle management (delete_tenant, delete_domain)
├── artifact_reader.py    # S3 artifact reading and parsing
├── artifact_validator.py # Schema compliance validation
├── graph_loader.py       # Neptune batch write with retry cascade
├── vector_loader.py      # OpenSearch Serverless vector ingestion
├── summary_overlay.py    # LLM summaries → Neptune node properties
├── code_slice_store.py   # Source evidence → Neptune node properties
└── graphson_converter.py # GraphSON format conversion

Integration

Architecture Stack

┌─────────────────────────────────────────────────────┐
│         codeproperty-graph (this package)            │
│  Joern/Semgrep CPG, delta ingestion, risk scoring   │
├─────────────────────────────────────────────────────┤
│              document-graph                          │
│  Node, Edge, CypherBuilder, PipelineExecutor        │
│  Multi-tenancy, batch operations                    │
├─────────────────────────────────────────────────────┤
│     graphrag-toolkit-lexical-graph (foundation)     │
│  GraphStore, Neptune writer, AOSS writer            │
│  Lexical indexing, entity resolution, retrieval     │
└─────────────────────────────────────────────────────┘

Delta Logic

1. Joern exports CPG → nodes.json + edges.json
2. Extract METHOD node signatures: {full_name: hash}
3. Compare against previous manifest in S3
4. If identical → SKIP (no Neptune writes, saves cost)
5. If changed → CPGService filters to changed methods only:
   • Identify added/modified METHOD nodes via GraphDiff.compare()
   • Walk AST edges to collect child nodes of changed methods
   • Filter edges to changed subgraph
   • Write only the delta (60-90% fewer Neptune writes)
   • Surgically remove deleted method subgraphs
   • Update manifest with full current state

With Document Graph

Code Property Graph uses document-graph for typed property graph primitives:

# document-graph provides the graph write infrastructure
from graphrag_toolkit.document_graph.graph_build.cypher_builder import CypherBuilder
from graphrag_toolkit.document_graph import Node, Edge

# codeproperty-graph adds CPG-specific semantics on top
from codeproperty_graph import CPGNode, CPGEdge

Contributing

See CONTRIBUTING.md for development setup, testing, and PR guidelines.

License

MIT — see LICENSE for details.

See NOTICE for third-party acknowledgments.