Spaces:
Sleeping
Sleeping
File size: 4,553 Bytes
6252f54 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | from typing import Annotated
from fastapi import APIRouter, Depends, Query
from backend.dependencies import get_neo4j_client
from backend.graph.neo4j_client import Neo4jClient
from backend.graph.cypher_queries import (
GET_ALL_DOMAINS,
GET_CAPABILITIES_FOR_DOMAIN,
GET_DOMAIN_STATS,
GET_DOMAIN_FULL_DETAIL,
)
router = APIRouter()
@router.get("/domains")
async def list_domains(neo4j: Annotated[Neo4jClient, Depends(get_neo4j_client)]):
rows = neo4j.run_query(GET_ALL_DOMAINS)
return [
{
"id": r["id"],
"name": r["name"],
"capability_count": r.get("capability_count", 0),
}
for r in rows
]
@router.get("/capabilities")
async def list_capabilities(
neo4j: Annotated[Neo4jClient, Depends(get_neo4j_client)],
domain_id: str | None = Query(default=None),
limit: int = Query(default=50, le=200),
):
rows = neo4j.run_query(GET_CAPABILITIES_FOR_DOMAIN, domain_id=domain_id, limit=limit)
return [
{
"id": r["id"],
"name": r["name"],
"domain": r.get("domain_name"),
"subdomain": r.get("subdomain_name"),
"description": r.get("description"),
}
for r in rows
]
@router.get("/stats")
async def graph_stats(neo4j: Annotated[Neo4jClient, Depends(get_neo4j_client)]):
rows = neo4j.run_query(GET_DOMAIN_STATS)
return {"node_counts": {r["label"]: r["count"] for r in rows}}
@router.get("/graph/network")
async def graph_network(neo4j: Annotated[Neo4jClient, Depends(get_neo4j_client)]):
from backend.graph.cypher_queries import GET_NETWORK_GRAPH
rows = neo4j.run_query(GET_NETWORK_GRAPH)
if not rows:
return {"nodes": [], "edges": []}
row = rows[0]
return {
"nodes": row.get("nodes") or [],
"edges": [e for e in (row.get("edges") or []) if e],
}
@router.get("/subdomains")
async def list_subdomains(
neo4j: Annotated[Neo4jClient, Depends(get_neo4j_client)],
domain_names: list[str] = Query(default=[]),
):
from backend.graph.cypher_queries import GET_SUBDOMAINS_FOR_DOMAINS
rows = neo4j.run_query(GET_SUBDOMAINS_FOR_DOMAINS, domain_names=domain_names)
return [
{
"id": r["id"],
"name": r["name"],
"domain_name": r.get("domain_name", ""),
"functional_scope": r.get("functional_scope", ""),
}
for r in rows
]
@router.get("/graph/domain-detail")
async def domain_detail(
domain_name: str,
neo4j: Annotated[Neo4jClient, Depends(get_neo4j_client)],
):
rows = neo4j.run_query(GET_DOMAIN_FULL_DETAIL, domain_name=domain_name)
if not rows:
return {"domain": {}, "standard": None, "trend": None, "subdomain_groups": []}
domain = rows[0].get("domain") or {}
standard = rows[0].get("standard")
trend = rows[0].get("trend")
subdomain_map: dict = {}
for r in rows:
sd = r.get("subdomain") or {}
sd_id = sd.get("id")
if not sd_id:
continue
if sd_id not in subdomain_map:
subdomain_map[sd_id] = {"subdomain": sd, "capabilities": []}
cap = r.get("capability") or {}
if cap.get("id"):
cap["subcapability_names"] = r.get("subcapability_names") or []
subdomain_map[sd_id]["capabilities"].append(cap)
# Sort subdomain groups and capabilities within each group alphabetically
groups = sorted(
subdomain_map.values(),
key=lambda g: (g["subdomain"].get("name") or "").lower(),
)
for g in groups:
g["capabilities"].sort(key=lambda c: (c.get("name") or "").lower())
return {
"domain": domain,
"standard": standard,
"trend": trend,
"subdomain_groups": groups,
}
@router.get("/subdomain-capabilities")
async def list_subdomain_capabilities(
neo4j: Annotated[Neo4jClient, Depends(get_neo4j_client)],
subdomain_ids: list[str] = Query(default=[]),
):
from backend.graph.cypher_queries import GET_CAPABILITIES_FOR_SUBDOMAINS
rows = neo4j.run_query(GET_CAPABILITIES_FOR_SUBDOMAINS, subdomain_ids=subdomain_ids)
return [
{
"id": r["id"],
"name": r["name"],
"subdomain": r.get("subdomain_name", ""),
"subdomain_id": r.get("subdomain_id", ""),
"description": r.get("description", ""),
"complexity": r.get("complexity", ""),
"duration_weeks": r.get("duration_weeks"),
}
for r in rows
]
|