"""Revisions API router - full CRUD for Kumiho revisions including tagging."""
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from app.dependencies import get_kumiho_client
from app.models.revision import (
RevisionCreate,
RevisionUpdate,
RevisionResponse,
TagRequest,
PeekRevisionResponse,
)
from app.models.common import parse_kref_params, StatusResponse
from app.core.kumiho_http import translate_kumiho_exception
from typing import Any
import kumiho
router = APIRouter()
[docs]
@router.get("", response_model=List[RevisionResponse])
async def list_revisions(
item_kref: str = Query(..., description="Item kref (e.g., 'MyProject/Assets/Hero.model')"),
client: Any = Depends(get_kumiho_client)
):
"""List all revisions of an item."""
try:
params = parse_kref_params(item_kref)
item = kumiho.get_item(params.kref_uri)
revisions = item.get_revisions()
return [RevisionResponse.from_domain(revision) for revision in revisions]
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.post("", response_model=RevisionResponse, status_code=status.HTTP_201_CREATED)
async def create_revision(
request: RevisionCreate,
client: Any = Depends(get_kumiho_client)
):
"""Create a new revision for an item."""
try:
params = parse_kref_params(request.item_kref)
item = kumiho.get_item(params.kref_uri)
revision = item.create_revision(
metadata=request.metadata,
number=request.number
)
return RevisionResponse.from_domain(revision)
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/by-kref", response_model=RevisionResponse)
async def get_revision(
kref: str = Query(..., description="Revision kref (e.g., 'MyProject/Assets/Hero.model?r=1')"),
r: Optional[int] = Query(None, description="Revision number"),
t: Optional[str] = Query(None, description="Tag to resolve (e.g., 'latest', 'published')"),
client: Any = Depends(get_kumiho_client)
):
"""Get a specific revision by kref with optional revision/tag resolution."""
try:
params = parse_kref_params(kref, r=r, t=t)
revision = kumiho.get_revision(params.kref_uri)
return RevisionResponse.from_domain(revision)
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/latest", response_model=RevisionResponse)
async def get_latest_revision(
item_kref: str = Query(..., description="Item kref"),
client: Any = Depends(get_kumiho_client)
):
"""Get the latest revision of an item."""
try:
params = parse_kref_params(item_kref)
item = kumiho.get_item(params.kref_uri)
revision = item.get_latest_revision()
if not revision:
raise HTTPException(status_code=404, detail="No revisions found")
return RevisionResponse.from_domain(revision)
except HTTPException:
raise
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/peek", response_model=PeekRevisionResponse)
async def peek_next_revision(
item_kref: str = Query(..., description="Item kref"),
client: Any = Depends(get_kumiho_client)
):
"""Get the next revision number that would be assigned."""
try:
params = parse_kref_params(item_kref)
item = kumiho.get_item(params.kref_uri)
next_number = item.peek_next_revision()
return PeekRevisionResponse(next_revision=next_number)
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.delete("/by-kref", status_code=status.HTTP_204_NO_CONTENT)
async def delete_revision(
kref: str = Query(..., description="Revision kref"),
r: Optional[int] = Query(None, description="Revision number"),
force: bool = Query(False, description="Force deletion"),
client: Any = Depends(get_kumiho_client)
):
"""Delete a revision."""
try:
params = parse_kref_params(kref, r=r)
revision = kumiho.get_revision(params.kref_uri)
revision.delete(force=force)
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.post("/deprecate", response_model=RevisionResponse)
async def deprecate_revision(
kref: str = Query(..., description="Revision kref"),
r: Optional[int] = Query(None, description="Revision number"),
deprecated: bool = Query(True, description="Deprecated status"),
client: Any = Depends(get_kumiho_client)
):
"""Deprecate or restore a revision."""
try:
params = parse_kref_params(kref, r=r)
revision = kumiho.get_revision(params.kref_uri)
revision.set_deprecated(deprecated)
return RevisionResponse.from_domain(revision)
except Exception as e:
raise translate_kumiho_exception(e)
# --- Tagging endpoints ---
[docs]
@router.post("/tags", response_model=StatusResponse)
async def tag_revision(
kref: str = Query(..., description="Revision kref"),
r: Optional[int] = Query(None, description="Revision number"),
request: Optional[TagRequest] = None,
client: Any = Depends(get_kumiho_client)
):
"""Apply a tag to a revision."""
try:
params = parse_kref_params(kref, r=r)
revision = kumiho.get_revision(params.kref_uri)
revision.tag(request.tag)
return StatusResponse(success=True, message=f"Tag '{request.tag}' applied")
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.delete("/tags", response_model=StatusResponse)
async def untag_revision(
kref: str = Query(..., description="Revision kref"),
tag: str = Query(..., description="Tag to remove"),
r: Optional[int] = Query(None, description="Revision number"),
client: Any = Depends(get_kumiho_client)
):
"""Remove a tag from a revision."""
try:
params = parse_kref_params(kref, r=r)
revision = kumiho.get_revision(params.kref_uri)
revision.untag(tag)
return StatusResponse(success=True, message=f"Tag '{tag}' removed")
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/tags/check")
async def has_tag(
kref: str = Query(..., description="Revision kref"),
tag: str = Query(..., description="Tag to check"),
r: Optional[int] = Query(None, description="Revision number"),
client: Any = Depends(get_kumiho_client)
):
"""Check if a revision has a specific tag."""
try:
params = parse_kref_params(kref, r=r)
revision = kumiho.get_revision(params.kref_uri)
has = revision.has_tag(tag)
return {"has_tag": has, "tag": tag}
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/tags/history")
async def was_tagged(
kref: str = Query(..., description="Revision kref"),
tag: str = Query(..., description="Tag to check"),
r: Optional[int] = Query(None, description="Revision number"),
client: Any = Depends(get_kumiho_client)
):
"""Check if a revision was ever tagged with a specific tag."""
try:
params = parse_kref_params(kref, r=r)
revision = kumiho.get_revision(params.kref_uri)
was = revision.was_tagged(tag)
return {"was_tagged": was, "tag": tag}
except Exception as e:
raise translate_kumiho_exception(e)
# --- Time-travel / As-Of endpoint ---
[docs]
@router.get("/as-of", response_model=RevisionResponse)
async def get_revision_as_of(
item_kref: str = Query(..., description="Item kref (e.g., 'MyProject/Assets/Hero.model')"),
tag: str = Query(..., description="Tag to query (e.g., 'published', 'approved')"),
time: str = Query(
...,
description="Timestamp in YYYYMMDDHHMM format (e.g., '202506011430') or ISO 8601 format (e.g., '2025-06-01T14:30:00Z')"
),
client: Any = Depends(get_kumiho_client)
):
"""Get the revision that had a specific tag at a given point in time.
This endpoint enables time-travel queries for reproducible builds and
historical analysis. It answers questions like:
"What was the published revision on June 1st, 2025?"
The query finds the revision that had the specified tag active at the
given timestamp, using the tag history time-ranges stored by Kumiho.
Args:
item_kref: The item reference (e.g., 'MyProject/Assets/Hero.model')
tag: The tag to query (commonly 'published', 'approved', 'latest')
time: The point in time to query. Supports two formats:
- YYYYMMDDHHMM: e.g., '202506011430' for June 1, 2025 at 14:30
- ISO 8601: e.g., '2025-06-01T14:30:00Z'
Returns:
RevisionResponse: The revision that had the tag at that time.
Raises:
404: If no revision had the specified tag at that time.
400: If the time format is invalid.
Example:
GET /api/v1/revisions/as-of?item_kref=MyProject/Assets/Hero.model&tag=published&time=202506011430
"""
try:
params = parse_kref_params(item_kref)
item = kumiho.get_item(params.kref_uri)
revision = item.get_revision_by_time(time, tag=tag)
if not revision:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"No revision with tag '{tag}' found at time '{time}'"
)
return RevisionResponse.from_domain(revision)
except HTTPException:
raise
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid time format: {str(e)}"
)
except Exception as e:
raise translate_kumiho_exception(e)