File size: 2,532 Bytes
345855e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | from fastapi import APIRouter
from fastapi.responses import JSONResponse
from typing import Dict, Any, List
from core.registry.tasks_registry import TASKS
# =====================================================
# INTERNAL HELPERS
# =====================================================
def _serialize_task(task) -> Dict[str, Any]:
"""
Convert TaskDefinition → API-safe metadata.
Must NOT import task modules.
"""
return {
"name": getattr(task, "name", None),
"description": getattr(task, "description", ""),
"category": getattr(task, "category", "general"),
"module": getattr(task, "module", None),
"callable": getattr(task, "callable_name", "run"),
"async": getattr(task, "async_task", False),
"enabled": getattr(task, "enabled", True),
}
def _group_tasks(tasks: List[Any]) -> Dict[str, List[Dict[str, Any]]]:
"""
Groups tasks by category for UI rendering.
"""
grouped: Dict[str, List[Dict[str, Any]]] = {}
for task in tasks:
category = getattr(task, "category", "general")
grouped.setdefault(category, []).append(_serialize_task(task))
return grouped
# =====================================================
# DOCS ROUTER BUILDER (V11)
# =====================================================
def build_docs_router() -> APIRouter:
"""
Builds dynamic documentation endpoints.
Provides:
/docs/tasks → flat task list
/docs/catalog → grouped tasks
/docs/health → docs status
"""
router = APIRouter(
prefix="/docs",
tags=["Documentation"],
)
# -------------------------------------------------
# List all tasks
# -------------------------------------------------
@router.get("/tasks")
async def list_tasks():
return JSONResponse(
[_serialize_task(task) for task in TASKS]
)
# -------------------------------------------------
# Categorized task catalog
# -------------------------------------------------
@router.get("/catalog")
async def task_catalog():
return JSONResponse(_group_tasks(TASKS))
# -------------------------------------------------
# Docs health endpoint
# -------------------------------------------------
@router.get("/health")
async def docs_health():
return {
"status": "ok",
"service": "docs",
"tasks_registered": len(TASKS),
}
return router |