"""Items API router - full CRUD for Kumiho items."""
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from app.dependencies import get_kumiho_client
from app.models.item import ItemCreate, ItemUpdate, ItemResponse, SearchResultResponse
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()
def _normalize_space_path(value: str) -> str:
path = (value or "").strip()
if not path:
return ""
# Normalize to a single leading slash and no trailing slash (except root).
path = "/" + path.strip("/")
return "/" if path == "//" else path.rstrip("/")
def _reject_template_like(value: str, *, field_name: str) -> None:
# Guard against accidentally passing n8n expression strings like
# `={{ $json.path }}` through as literal values.
s = (value or "").strip()
if not s:
return
if "{{" in s or "}}" in s or s.startswith("={{") or s.startswith("="):
raise HTTPException(
status_code=400,
detail=(
f"{field_name} looks like an unevaluated expression. "
"Provide a literal Kumiho path like '/MyProject/Assets'."
),
)
[docs]
@router.get("", response_model=List[ItemResponse])
async def list_items(
space_path: str = Query(..., description="Space path (e.g., '/MyProject/Assets')"),
name_filter: Optional[str] = Query(None, description="Filter by item name"),
kind_filter: Optional[str] = Query(None, description="Filter by item kind"),
client: Any = Depends(get_kumiho_client)
):
"""List items in a specific space."""
try:
space_path = _normalize_space_path(space_path)
_reject_template_like(space_path, field_name="space_path")
if not space_path or space_path == "/":
raise HTTPException(status_code=400, detail="space_path must include a project segment")
project_name = space_path.strip('/').split('/')[0]
project = kumiho.get_project(project_name)
if not project:
raise HTTPException(status_code=404, detail="Project not found")
space = project.get_space(space_path)
items = space.get_items(
item_name_filter=name_filter or "",
kind_filter=kind_filter or ""
)
return [ItemResponse.from_domain(item) for item in items]
except HTTPException:
raise
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/search", response_model=List[ItemResponse])
async def search_items(
context_filter: Optional[str] = Query(None, description="Filter by context/path"),
name_filter: Optional[str] = Query(None, description="Filter by item name"),
kind_filter: Optional[str] = Query(None, description="Filter by item kind"),
client: Any = Depends(get_kumiho_client)
):
"""Search for items across the system using pattern/glob matching."""
try:
items = kumiho.item_search(
context_filter=context_filter or "",
name_filter=name_filter or "",
kind_filter=kind_filter or ""
)
return [ItemResponse.from_domain(item) for item in items]
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/fulltext-search", response_model=List[SearchResultResponse])
async def fulltext_search(
query: str = Query(..., description="Search query (supports fuzzy matching)"),
context: Optional[str] = Query(None, description="Restrict to kref prefix (e.g., 'myproject/assets')"),
kind: Optional[str] = Query(None, description="Exact kind match (e.g., 'model', 'texture')"),
include_deprecated: bool = Query(False, description="Include soft-deleted items"),
include_revision_metadata: bool = Query(
False, description="Also search revision tags/metadata (slower but more comprehensive)"
),
include_artifact_metadata: bool = Query(
False, description="Also search artifact names/metadata (slower)"
),
client: Any = Depends(get_kumiho_client)
):
"""Full-text fuzzy search across items (Google-like search).
Provides Google-like search with automatic typo tolerance. Searches
across item names, kinds, usernames, and optionally revision/artifact
metadata. Results are ranked by relevance score.
"""
try:
results = kumiho.search(
query,
context=context or "",
kind=kind or "",
include_deprecated=include_deprecated,
include_revision_metadata=include_revision_metadata,
include_artifact_metadata=include_artifact_metadata,
)
return [SearchResultResponse.from_domain(r) for r in results]
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.post("", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(
request: ItemCreate,
client: Any = Depends(get_kumiho_client)
):
"""Create a new item."""
try:
space_path = _normalize_space_path(request.space_path)
_reject_template_like(request.space_path, field_name="space_path")
if not space_path or space_path == "/":
raise HTTPException(status_code=400, detail="space_path must include a project segment")
project_name = space_path.strip('/').split('/')[0]
project = kumiho.get_project(project_name)
if not project:
raise HTTPException(status_code=404, detail="Project not found")
item = project.create_item(
item_name=request.item_name,
kind=request.kind,
parent_path=space_path,
metadata=request.metadata or None,
)
return ItemResponse.from_domain(item)
except HTTPException:
raise
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/by-kref", response_model=ItemResponse)
async def get_item_by_kref(
kref: str = Query(..., description="Item kref (e.g., 'MyProject/Assets/Hero.model')"),
client: Any = Depends(get_kumiho_client)
):
"""Get an item by its kref."""
try:
params = parse_kref_params(kref)
item = kumiho.get_item(params.kref_uri)
return ItemResponse.from_domain(item)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/by-path", response_model=ItemResponse)
async def get_item_by_path(
space_path: str = Query(..., description="Space path"),
item_name: str = Query(..., description="Item name"),
kind: str = Query(..., description="Item kind"),
client: Any = Depends(get_kumiho_client)
):
"""Get an item by its path components."""
try:
space_path = _normalize_space_path(space_path)
_reject_template_like(space_path, field_name="space_path")
if not space_path or space_path == "/":
raise HTTPException(status_code=400, detail="space_path must include a project segment")
project_name = space_path.strip('/').split('/')[0]
project = kumiho.get_project(project_name)
if not project:
raise HTTPException(status_code=404, detail="Project not found")
space = project.get_space(space_path)
item = space.get_item(
item_name=item_name,
kind=kind
)
return ItemResponse.from_domain(item)
except HTTPException:
raise
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.delete("/by-kref", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(
kref: str = Query(..., description="Item kref"),
force: bool = Query(False, description="Force deletion"),
client: Any = Depends(get_kumiho_client)
):
"""Delete an item."""
try:
params = parse_kref_params(kref)
item = kumiho.get_item(params.kref_uri)
item.delete(force=force)
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.post("/deprecate", response_model=ItemResponse)
async def deprecate_item(
kref: str = Query(..., description="Item kref"),
deprecated: bool = Query(True, description="Deprecated status"),
client: Any = Depends(get_kumiho_client)
):
"""Deprecate or restore an item."""
try:
params = parse_kref_params(kref)
item = kumiho.get_item(params.kref_uri)
item.set_deprecated(deprecated)
return ItemResponse.from_domain(item)
except Exception as e:
raise translate_kumiho_exception(e)