"""Resolve API router - kref resolution utilities."""
from typing import Any, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from app.dependencies import get_kumiho_client
from app.models.common import parse_kref_params
from app.models.revision import RevisionResponse
import kumiho
router = APIRouter()
[docs]
class ResolveResponse(BaseModel):
"""Response for kref resolution."""
kref: str
location: Optional[str] = None
resolved_revision: Optional[int] = None
resolved_artifact: Optional[str] = None
[docs]
class ResolveRevisionResponse(BaseModel):
"""Full kref resolution response with revision details."""
revision: Optional[RevisionResponse] = None
found: bool = False
[docs]
@router.get("", response_model=ResolveResponse)
async def resolve_kref(
kref: str = Query(..., description="Kref to resolve (e.g., 'MyProject/Assets/Hero.model')"),
r: Optional[int] = Query(None, description="Revision number"),
t: Optional[str] = Query(None, description="Tag to resolve (e.g., 'latest', 'published')"),
a: Optional[str] = Query(None, description="Artifact name"),
time: Optional[str] = Query(None, description="Time in YYYYMMDDHHMM format"),
client: Any = Depends(get_kumiho_client)
):
"""
Resolve a kref to a file location.
Resolution logic:
- Item kref → latest revision → default artifact → location
- Revision kref → default artifact → location
- Artifact kref → location directly
Query parameters can override embedded values in the kref.
"""
try:
params = parse_kref_params(kref, r=r, t=t, a=a, time=time)
location = kumiho.resolve(params.kref_uri)
# `parse_kref_params()` only reflects what the caller provided.
# When the caller omits `a=...`, `kumiho.resolve()` can still resolve
# via the revision's default artifact. In that case, populate
# resolved_revision/resolved_artifact from the resolved revision.
resolved_revision = params.revision
resolved_artifact = params.artifact
if resolved_revision is None or resolved_artifact is None:
try:
revision = kumiho.get_revision(params.kref_uri)
if resolved_revision is None:
resolved_revision = getattr(revision, "number", None)
if resolved_artifact is None:
resolved_artifact = getattr(revision, "default_artifact", None)
except Exception:
# Best-effort enrichment only; keep the original fields.
pass
return ResolveResponse(
kref=params.kref_uri,
location=location,
resolved_revision=resolved_revision,
resolved_artifact=resolved_artifact,
)
except Exception as e:
if "not found" in str(e).lower():
raise HTTPException(status_code=404, detail="Kref could not be resolved")
raise HTTPException(status_code=500, detail=str(e))
[docs]
@router.get("/revision", response_model=ResolveRevisionResponse)
async def resolve_kref_to_revision(
kref: str = Query(..., description="Kref to resolve"),
t: Optional[str] = Query(None, description="Tag to resolve"),
time: Optional[str] = Query(None, description="Time in YYYYMMDDHHMM format"),
client: Any = Depends(get_kumiho_client)
):
"""
Resolve a kref to a specific revision with optional tag/time constraints.
This is useful for point-in-time queries or finding which revision
had a specific tag at a given moment.
"""
try:
params = parse_kref_params(kref, t=t, time=time)
revision = kumiho.get_revision(params.kref_uri)
return ResolveRevisionResponse(
revision=RevisionResponse.from_domain(revision),
found=True
)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
if "not found" in str(e).lower():
return ResolveRevisionResponse(found=False)
raise HTTPException(status_code=500, detail=str(e))