Ownership Methods

def get_owner(node_id: str) -> Optional[Dict[str, Any]]

from graph.query import QueryEngine

query_engine = QueryEngine(storage)

# Find who owns a service
owner = query_engine.get_owner("service:payment-api")

if owner:
    print(f"Owner: {owner['name']}")
    if 'slack_channel' in owner:
        print(f"Contact: {owner['slack_channel']}")
else:
    print("No owner assigned")

# During an incident, find the responsible team
failing_service = "service:auth-service"
owner = query_engine.get_owner(failing_service)

if owner:
    print(f"🚨 INCIDENT ALERT")
    print(f"Service: {failing_service}")
    print(f"Responsible Team: {owner['name']}")
    
    # Get contact information
    if 'slack_channel' in owner:
        print(f"📱 Alert: {owner['slack_channel']}")
    if 'pagerduty_schedule' in owner:
        print(f"📟 Page: {owner['pagerduty_schedule']}")
else:
    print("⚠️  No owner found - escalating to platform team")

# Check ownership coverage
services = query_engine.get_nodes(node_type="service")

unowned = []
owned_by_team = {}

for service in services:
    owner = query_engine.get_owner(service['id'])
    
    if owner:
        team_name = owner['name']
        if team_name not in owned_by_team:
            owned_by_team[team_name] = []
        owned_by_team[team_name].append(service['name'])
    else:
        unowned.append(service['name'])

print(f"Ownership Coverage: {len(services) - len(unowned)}/{len(services)}")
print(f"\n⚠️  Unowned services ({len(unowned)}):")
for name in unowned:
    print(f"  - {name}")

print(f"\n✓ Owned services by team:")
for team, services in owned_by_team.items():
    print(f"  {team}: {len(services)} services")

def route_issue(component_id: str, issue_description: str):
    """Route an issue to the appropriate team."""
    owner = query_engine.get_owner(component_id)
    
    if owner:
        return {
            'team': owner['name'],
            'contact': owner.get('slack_channel', 'No contact info'),
            'issue': issue_description,
            'component': component_id
        }
    else:
        return {
            'team': 'Platform',  # Default escalation
            'contact': '#platform-team',
            'issue': issue_description,
            'component': component_id,
            'note': 'No owner found - defaulting to platform team'
        }

# Use it
routing = route_issue(
    "database:users-db",
    "High connection count"
)

print(f"Route to: {routing['team']}")
print(f"Contact: {routing['contact']}")

def get_team_assets(team_name: str) -> List[Dict[str, Any]]

from graph.query import QueryEngine

query_engine = QueryEngine(storage)

# List all assets owned by a team
assets = query_engine.get_team_assets("Platform")

print(f"Platform team owns {len(assets)} assets:\n")

for asset in assets:
    print(f"  [{asset['type']}] {asset['name']}")

Platform team owns 12 assets:

  [database] postgres-main
  [database] redis-cache
  [library] auth-lib
  [library] logging-lib
  [service] api-gateway
  [service] auth-service
  [service] config-service
  ...

# Generate a detailed inventory for a team
team_name = "Identity"
assets = query_engine.get_team_assets(team_name)

# Group by type
by_type = {}
for asset in assets:
    asset_type = asset['type']
    if asset_type not in by_type:
        by_type[asset_type] = []
    by_type[asset_type].append(asset)

print(f"=== {team_name} Team Asset Inventory ===")
print(f"\nTotal Assets: {len(assets)}\n")

for asset_type, items in sorted(by_type.items()):
    print(f"{asset_type.upper()} ({len(items)}):")
    for item in items:
        print(f"  - {item['name']}")
        if 'language' in item:
            print(f"    Language: {item['language']}")
        if 'version' in item:
            print(f"    Version: {item['version']}")
    print()

# Compare assets across teams
teams = ["Platform", "Identity", "Payments", "API"]

print("Team Responsibility Matrix:\n")
print(f"{'Team':<15} {'Services':<10} {'Databases':<12} {'Libraries':<10} {'Total':<8}")
print("-" * 60)

for team_name in teams:
    assets = query_engine.get_team_assets(team_name)
    
    services = sum(1 for a in assets if a['type'] == 'service')
    databases = sum(1 for a in assets if a['type'] == 'database')
    libraries = sum(1 for a in assets if a['type'] == 'library')
    
    print(f"{team_name:<15} {services:<10} {databases:<12} {libraries:<10} {len(assets):<8}")

# Find total impact if entire team is unavailable
team_name = "Platform"
assets = query_engine.get_team_assets(team_name)

print(f"Impact Analysis: {team_name} Team Unavailable\n")

total_upstream = 0
affected_teams = set()

for asset in assets:
    # Get upstream dependencies for each asset
    upstream = query_engine.upstream(asset['id'], max_depth=5)
    total_upstream += len(upstream)
    
    # Find teams affected
    for dep in upstream:
        owner = query_engine.get_owner(dep['id'])
        if owner and owner['name'] != team_name:
            affected_teams.add(owner['name'])

print(f"Team owns: {len(assets)} assets")
print(f"Total upstream dependencies: {total_upstream}")
print(f"\nOther teams affected: {len(affected_teams)}")
for team in sorted(affected_teams):
    print(f"  - {team}")

# Generate onboarding guide for new team member
team_name = "API"
assets = query_engine.get_team_assets(team_name)

print(f"=== Welcome to {team_name} Team! ===")
print(f"\nYou'll be responsible for {len(assets)} components:\n")

for asset in assets:
    print(f"\n📦 {asset['name']} ({asset['type']})")
    print(f"   ID: {asset['id']}")
    
    # Show key dependencies
    deps = query_engine.downstream(asset['id'], max_depth=2)
    if deps:
        print(f"   Key dependencies: {', '.join([d['name'] for d in deps[:3]])}")
    
    # Show main consumers
    consumers = query_engine.upstream(asset['id'], max_depth=1)
    if consumers:
        print(f"   Used by: {', '.join([c['name'] for c in consumers[:3]])}")

# Find assets with multiple owners (should not happen)
all_nodes = query_engine.get_nodes()

multiple_owners = []

for node in all_nodes:
    # This query finds ALL teams that own a node
    query = """
    MATCH (team {type: 'team'})-[:OWNS]->(node {id: $node_id})
    RETURN team
    """
    owners = query_engine.storage.execute_cypher(query, {'node_id': node['id']})
    
    if len(owners) > 1:
        multiple_owners.append({
            'asset': node['name'],
            'owners': [dict(o['team'])['name'] for o in owners]
        })

if multiple_owners:
    print(f"⚠️  Found {len(multiple_owners)} assets with multiple owners:")
    for item in multiple_owners:
        print(f"  {item['asset']}: {', '.join(item['owners'])}")
else:
    print("✓ No ownership conflicts found")