"""Spaces API router - full CRUD for Kumiho spaces."""
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from app.dependencies import get_kumiho_client
from app.models.space import SpaceCreate, SpaceUpdate, SpaceResponse
from app.core.kumiho_http import translate_kumiho_exception
from typing import Any
import kumiho
router = APIRouter()
[docs]
@router.get("", response_model=List[SpaceResponse])
async def list_spaces(
parent_path: str = Query(..., description="Parent path (e.g., '/MyProject' or '/MyProject/Assets')"),
recursive: bool = Query(False, description="Fetch all descendant spaces recursively"),
client: Any = Depends(get_kumiho_client)
):
"""List child spaces under a parent path."""
try:
project_name = parent_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(parent_path)
spaces = space.get_spaces(recursive=recursive)
return [SpaceResponse.from_domain(s) for s in spaces]
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.post("", response_model=SpaceResponse, status_code=status.HTTP_201_CREATED)
async def create_space(
request: SpaceCreate,
client: Any = Depends(get_kumiho_client)
):
"""Create a new space."""
try:
parent_path = (request.parent_path or "").strip()
if not parent_path or parent_path == "/":
raise HTTPException(
status_code=400,
detail="parent_path must include a project segment (e.g. '/MyProject' or '/MyProject/Assets')",
)
project_name = parent_path.strip('/').split('/')[0]
project = kumiho.get_project(project_name)
if not project:
raise HTTPException(status_code=404, detail="Project not found")
# Ensure the project root space exists (/{project_name}).
root_path = f"/{project_name}"
try:
project.get_space(root_path)
except Exception:
# Create the root space under '/' (idempotent on the server side).
project.create_space(name=project_name, parent_path="/")
# Ensure the full parent_path hierarchy exists (create missing intermediate spaces).
# Example: parent_path='/proj/Assets/Characters' -> create Assets then Characters.
parent_segments = [p for p in parent_path.strip('/').split('/') if p]
current_space = project.get_space(root_path)
for seg in parent_segments[1:]:
try:
current_space = current_space.get_space(seg)
except Exception:
current_space = current_space.create_space(seg)
# Create the requested child space, but treat "already exists" as idempotent.
try:
space = current_space.create_space(name=request.name)
except Exception:
space = current_space.get_space(request.name)
return SpaceResponse.from_domain(space)
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.get("/by-path", response_model=SpaceResponse)
async def get_space(
path: str = Query(..., description="Full space path (e.g., '/MyProject/Assets/Characters')"),
client: Any = Depends(get_kumiho_client)
):
"""Get a specific space by its path."""
try:
project_name = 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(path)
return SpaceResponse.from_domain(space)
except Exception as e:
raise translate_kumiho_exception(e)
[docs]
@router.delete("/by-path", status_code=status.HTTP_204_NO_CONTENT)
async def delete_space(
path: str = Query(..., description="Full space path"),
force: bool = Query(False, description="Force deletion even if space has content"),
client: Any = Depends(get_kumiho_client)
):
"""Delete a space."""
try:
project_name = 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(path)
space.delete(force=force)
except Exception as e:
raise translate_kumiho_exception(e)