xsl
746b66da01
One request exercises every layer: JWT auth, quota reservation, a versioned prompt through the LLM gateway, a cost row, quota commit or refund, and the unified envelope. The trivial ping task proves the queue routing and the tk_task transitions before any real long-running task is written. Routes are mounted at the /api/v1 base path of 架构文档 §9.1; /v1/* is rewritten to it at the ASGI layer so both spellings work from one route table.
81 lines
2.6 KiB
Python
Executable File
81 lines
2.6 KiB
Python
Executable File
"""任务中心 endpoints + a trivial async task that proves the Celery path.
|
|
|
|
The ``ping`` task carries no business value; it exists so that the ``tk_task``
|
|
state machine, the queue routing and the worker deployment are all exercised
|
|
before any real long-running task is written (M1 出口条件).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Query
|
|
|
|
from apps.api.deps import CurrentUserId, TaskServiceDep
|
|
from apps.api.schemas import TaskResponse, TaskSubmitRequest
|
|
from packages.domain.task.state import TaskStatus, TaskType
|
|
from packages.shared.models import Task
|
|
from packages.shared.response import ok
|
|
|
|
router = APIRouter(prefix="/tasks", tags=["tasks"])
|
|
|
|
|
|
def _serialize(task: Task) -> dict[str, object]:
|
|
return TaskResponse(
|
|
task_id=str(task.task_id),
|
|
task_type=task.task_type,
|
|
status=task.status,
|
|
progress=task.progress,
|
|
stage_text=task.stage_text,
|
|
result=task.result,
|
|
error_code=task.error_code,
|
|
error_message=task.error_message,
|
|
retry_count=task.retry_count,
|
|
submitted_at=task.submitted_at,
|
|
started_at=task.started_at,
|
|
finished_at=task.finished_at,
|
|
).model_dump()
|
|
|
|
|
|
@router.post("/ping", summary="Submit the scaffold's trivial async task")
|
|
async def submit_ping(
|
|
payload: TaskSubmitRequest,
|
|
user_id: CurrentUserId,
|
|
tasks: TaskServiceDep,
|
|
) -> dict[str, object]:
|
|
from apps.worker.tasks import ping_task
|
|
|
|
idempotency_key = payload.idempotency_key or f"ping:{user_id}:{uuid.uuid4().hex}"
|
|
task = await tasks.submit(
|
|
user_id=user_id,
|
|
task_type=TaskType.PING,
|
|
idempotency_key=idempotency_key,
|
|
payload={"echo": payload.echo},
|
|
eta_seconds=2,
|
|
)
|
|
if task.status == TaskStatus.PENDING.value:
|
|
task = await tasks.transition(
|
|
task_id=task.task_id, target=TaskStatus.QUEUED, stage_text="排队中"
|
|
)
|
|
ping_task.apply_async(args=[task.task_id], queue="cpu")
|
|
return ok(_serialize(task))
|
|
|
|
|
|
@router.get("/{task_id}", summary="Task status (ownership checked)")
|
|
async def get_task(
|
|
task_id: int,
|
|
user_id: CurrentUserId,
|
|
tasks: TaskServiceDep,
|
|
) -> dict[str, object]:
|
|
return ok(_serialize(await tasks.get(task_id=task_id, user_id=user_id)))
|
|
|
|
|
|
@router.get("", summary="Task centre list")
|
|
async def list_tasks(
|
|
user_id: CurrentUserId,
|
|
tasks: TaskServiceDep,
|
|
limit: int = Query(default=20, ge=1, le=100),
|
|
) -> dict[str, object]:
|
|
rows = await tasks.list_for_user(user_id=user_id, limit=limit)
|
|
return ok([_serialize(row) for row in rows])
|