Spaces:
Sleeping
Sleeping
File size: 5,187 Bytes
c045254 | 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 | """Knowledge graph API endpoints."""
import logging
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import JSONResponse
from app.core.exceptions import EntityNotFoundError, DatasetError
from app.schemas.knowledge_graph import KnowledgeGraphResponse
from app.services.knowledge_graph_service import KnowledgeGraphService
from app.api.v2.dependencies import get_knowledge_graph_service
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/ai/v2/knowledge-graph", tags=["knowledge-graph"])
# NOTE: Static-segment routes MUST be declared before dynamic /{lo_id} routes
# so FastAPI doesn't swallow "path" as a lo_id value.
@router.get(
"/path/{start_lo}/{target_lo}",
summary="Get shortest path between two learning outcomes",
description="Find the shortest dependency path between two learning outcomes in the knowledge graph"
)
async def get_learning_path_between_los(
start_lo: str,
target_lo: str,
knowledge_graph_service: KnowledgeGraphService = Depends(get_knowledge_graph_service)
) -> JSONResponse:
"""Get learning path between two learning outcomes."""
try:
path = knowledge_graph_service.get_learning_path(start_lo, target_lo)
return JSONResponse(content={
"start_lo": start_lo,
"target_lo": target_lo,
"path": path,
"steps": len(path)
})
except EntityNotFoundError as exc:
logger.warning("Learning path query failed - entity not found: %s", exc)
raise HTTPException(status_code=404, detail=str(exc)) from exc
except Exception as exc:
logger.error("Learning path query failed: %s", exc)
raise HTTPException(status_code=500, detail="Internal server error") from exc
@router.get(
"/{lo_id}/prerequisites",
summary="Get prerequisites for a learning outcome",
description="Get all prerequisite learning outcomes for a given LO"
)
async def get_prerequisites(
lo_id: str,
max_depth: int = Query(default=None, ge=1, le=10, description="Maximum depth to traverse"),
knowledge_graph_service: KnowledgeGraphService = Depends(get_knowledge_graph_service)
) -> JSONResponse:
"""Get prerequisites for a learning outcome."""
try:
prerequisites = knowledge_graph_service.get_prerequisites(lo_id, max_depth)
return JSONResponse(content={
"lo_id": lo_id,
"prerequisites": prerequisites,
"count": len(prerequisites)
})
except EntityNotFoundError as exc:
logger.warning("Prerequisites query failed - entity not found: %s", exc)
raise HTTPException(status_code=404, detail=str(exc)) from exc
except Exception as exc:
logger.error("Prerequisites query failed: %s", exc)
raise HTTPException(status_code=500, detail="Internal server error") from exc
@router.get(
"/{lo_id}/successors",
summary="Get successors for a learning outcome",
description="Get all successor learning outcomes for a given LO"
)
async def get_successors(
lo_id: str,
max_depth: int = Query(default=None, ge=1, le=10, description="Maximum depth to traverse"),
knowledge_graph_service: KnowledgeGraphService = Depends(get_knowledge_graph_service)
) -> JSONResponse:
"""Get successors for a learning outcome."""
try:
successors = knowledge_graph_service.get_successors(lo_id, max_depth)
return JSONResponse(content={
"lo_id": lo_id,
"successors": successors,
"count": len(successors)
})
except EntityNotFoundError as exc:
logger.warning("Successors query failed - entity not found: %s", exc)
raise HTTPException(status_code=404, detail=str(exc)) from exc
except Exception as exc:
logger.error("Successors query failed: %s", exc)
raise HTTPException(status_code=500, detail="Internal server error") from exc
@router.get(
"/{lo_id}",
response_model=KnowledgeGraphResponse,
summary="Get knowledge graph for a learning outcome",
description="Retrieve full knowledge graph info including prerequisites and successors for a learning outcome"
)
async def get_knowledge_graph(
lo_id: str,
max_depth: int = Query(default=2, ge=1, le=5, description="Maximum depth to traverse"),
knowledge_graph_service: KnowledgeGraphService = Depends(get_knowledge_graph_service)
) -> KnowledgeGraphResponse:
"""Get knowledge graph information for a learning outcome."""
try:
return knowledge_graph_service.get_knowledge_graph(lo_id, max_depth)
except EntityNotFoundError as exc:
logger.warning("Knowledge graph query failed - entity not found: %s", exc)
raise HTTPException(status_code=404, detail=str(exc)) from exc
except DatasetError as exc:
logger.error("Knowledge graph query failed - dataset error: %s", exc)
raise HTTPException(status_code=503, detail="Knowledge graph service unavailable") from exc
except Exception as exc:
logger.error("Knowledge graph query failed - unexpected error: %s", exc)
raise HTTPException(status_code=500, detail="Internal server error") from exc
|