blast_radius()

def blast_radius(
    node_id: str,
    max_depth: int = 10
) -> Dict[str, Any]

from graph.query import QueryEngine

query_engine = QueryEngine(storage)

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

print(impact['summary'])
print(f"\nAffected teams: {', '.join(impact['affected_teams'])}")
print(f"Total affected nodes: {impact['total_affected_nodes']}")

If users-db fails, it could affect 12 upstream and 3 downstream components, impacting 5 teams.

Affected teams: Platform, Identity, Payments, API, Frontend
Total affected nodes: 16

# Before deploying a breaking change
service_id = "service:auth-service"
impact = query_engine.blast_radius(service_id)

if impact['total_affected_nodes'] > 10:
    print("⚠️  HIGH RISK DEPLOYMENT")
    print(f"   Affects {len(impact['affected_teams'])} teams")
    print(f"   Impacts {len(impact['upstream_dependencies'])} services")
    
    # Notify affected teams
    print("\n📧 Teams to notify:")
    for team in impact['affected_teams']:
        print(f"   - {team}")
else:
    print("✓ Low risk deployment")

impact = query_engine.blast_radius("service:payment-api")

print(f"=== Blast Radius Analysis for {impact['center_node']['name']} ===")
print(f"\nNode Type: {impact['center_node']['type']}")

print(f"\n📤 UPSTREAM IMPACT ({len(impact['upstream_dependencies'])} nodes)")
print("   Services that will break if this fails:")
for node in impact['upstream_dependencies'][:5]:  # Show first 5
    print(f"   - {node['name']} ({node['type']}) [distance: {node['distance']}]")

print(f"\n📥 DOWNSTREAM DEPENDENCIES ({len(impact['downstream_dependencies'])} nodes)")
print("   Components this service needs:")
for node in impact['downstream_dependencies'][:5]:  # Show first 5
    print(f"   - {node['name']} ({node['type']}) [distance: {node['distance']}]")

print(f"\n👥 AFFECTED TEAMS ({len(impact['affected_teams'])} teams)")
for team in impact['affected_teams']:
    print(f"   - {team}")

print(f"\n📊 SUMMARY")
print(f"   {impact['summary']}")

# During an outage
failed_service = "database:postgres-main"
impact = query_engine.blast_radius(failed_service)

print("🚨 INCIDENT IMPACT REPORT")
print(f"\nFailed component: {impact['center_node']['name']}")
print(f"\n⚠️  {len(impact['upstream_dependencies'])} services are likely DOWN")

# Group affected services by team
from collections import defaultdict
team_services = defaultdict(list)

for node in impact['upstream_dependencies']:
    # Get owner for each affected node
    owner = query_engine.get_owner(node['id'])
    if owner:
        team_services[owner['name']].append(node['name'])

print("\nImpact by team:")
for team, services in team_services.items():
    print(f"\n  {team} ({len(services)} services):")
    for service in services:
        print(f"    - {service}")

# Compare blast radius of different databases
databases = [
    "database:users-db",
    "database:analytics-db",
    "database:cache-db"
]

print("Database Impact Comparison:\n")
for db_id in databases:
    impact = query_engine.blast_radius(db_id, max_depth=5)
    print(f"{impact['center_node']['name']}:")
    print(f"  Upstream: {len(impact['upstream_dependencies'])} nodes")
    print(f"  Downstream: {len(impact['downstream_dependencies'])} nodes")
    print(f"  Teams: {len(impact['affected_teams'])}")
    print()