QueryEngine

​

Overview

​

Initialization

from graph.query import QueryEngine
from graph.storage import GraphStorage

# Initialize storage and query engine
storage = GraphStorage(uri="neo4j://localhost:7687")
query_engine = QueryEngine(storage)

​

Core Capabilities

​

Dependency Analysis

• Downstream dependencies: Find what a node depends on
• Upstream dependencies: Find what depends on a node
• Blast radius: Full impact analysis including affected teams

​

Path Finding

• Shortest path: Find connections between any two nodes

​

Ownership

• Get owner: Find the team that owns an asset
• Get team assets: List all assets owned by a team

​

Basic Usage

# Get downstream dependencies
deps = query_engine.downstream(
    node_id="service:payment-api",
    max_depth=5
)

# Find impact of a change
impact = query_engine.blast_radius(
    node_id="database:users-db",
    max_depth=10
)

# Find ownership
owner = query_engine.get_owner(node_id="service:auth-service")

​

Method Reference

​

Return Types

• Node lists return List[Dict[str, Any]]
• Single nodes return Optional[Dict[str, Any]]
• Complex results return Dict[str, Any] with structured data

​

Performance Considerations

• Use max_depth to limit traversal depth and prevent performance issues
• Filter by edge_types to narrow results and improve query speed
• Large graphs may require tuning of depth parameters