"""Edges API router - full CRUD for Kumiho edges."""
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from app.dependencies import get_kumiho_client
from app.models.edge import EdgeCreate, EdgeResponse
from app.models.common import parse_kref_params
from app.core.kumiho_http import translate_kumiho_exception
from typing import Any
import kumiho
router = APIRouter()
[docs]
@router.get("", response_model=List[EdgeResponse])
async def list_edges(
kref: str = Query(..., description="Revision kref to get edges for"),
r: Optional[int] = Query(None, description="Revision number"),
edge_type: Optional[str] = Query(None, description="Filter by edge type"),
direction: Optional[str] = Query(None, description="Filter by edge direction (outgoing, incoming, both)"),
client: Any = Depends(get_kumiho_client)
):
"""List all edges for a revision."""
try:
params = parse_kref_params(kref, r=r)
revision = kumiho.get_revision(params.kref_uri)
direction_map = {
"outgoing": kumiho.OUTGOING,
"incoming": kumiho.INCOMING,
"both": kumiho.BOTH,
"source": kumiho.OUTGOING,
"target": kumiho.INCOMING,
}
direction_value = (
direction_map.get(direction.lower(), kumiho.BOTH)
if direction
else kumiho.BOTH
)
edges = revision.get_edges(edge_type_filter=edge_type, direction=direction_value)
return [EdgeResponse.from_domain(edge) for edge in edges]
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.post("", response_model=EdgeResponse, status_code=status.HTTP_201_CREATED)
async def create_edge(
request: EdgeCreate,
client: Any = Depends(get_kumiho_client)
):
"""Create an edge between two revisions."""
try:
# Get the source and target revisions
source_params = parse_kref_params(request.source_revision_kref)
target_params = parse_kref_params(request.target_revision_kref)
source_revision = kumiho.get_revision(source_params.kref_uri)
target_revision = kumiho.get_revision(target_params.kref_uri)
edge = source_revision.create_edge(
target_revision=target_revision,
edge_type=request.edge_type,
metadata=request.metadata
)
return EdgeResponse.from_domain(edge)
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.delete("", status_code=status.HTTP_204_NO_CONTENT)
async def delete_edge(
source_kref: str = Query(..., description="Source revision kref"),
target_kref: str = Query(..., description="Target revision kref"),
edge_type: str = Query(..., description="Type of edge to delete"),
client: Any = Depends(get_kumiho_client)
):
"""Delete an edge between two revisions."""
try:
source_params = parse_kref_params(source_kref)
target_params = parse_kref_params(target_kref)
source_revision = kumiho.get_revision(source_params.kref_uri)
target_revision = kumiho.get_revision(target_params.kref_uri)
source_revision.delete_edge(
target_revision=target_revision,
edge_type=edge_type
)
except Exception as e:
raise translate_kumiho_exception(e)