taolib.testing.task_queue.server.api.tasks

taolib.testing.task_queue.server.api.tasks#

任务路由。

Attributes#

router

TASKSCHEDULE_MODULE_DESCRIPTION

Classes#

TaskListResponse

任务列表响应。

SubmitTaskRequest

提交任务请求。

TaskActionResponse

任务操作响应。

Functions#

get_task_service(request)

获取任务服务。

list_tasks(→ TaskListResponse)

列出任务。

get_task(→ taolib.testing.task_queue.models.TaskResponse)

获取任务详情。

submit_task(...)

提交新任务。

retry_task(→ taolib.testing.task_queue.models.TaskResponse)

手动重试失败任务。

cancel_task(...)

取消任务。

delete_task(→ None)

删除终态任务（仅 COMPLETED/FAILED/CANCELLED）。

Module Contents#

taolib.testing.task_queue.server.api.tasks.router#
taolib.testing.task_queue.server.api.tasks.TASKSCHEDULE_MODULE_DESCRIPTION = Multiline-String#
Show Value
"""
任务队列 API 文档
任务队列 API 提供任务提交、查询和管理功能。

## 功能特性

- **任务提交**：支持异步任务提交和执行
- **优先级队列**：高/普通/低优先级
- **失败重试**：指数退避重试
- **幂等性**：支持幂等任务
- **任务标签**：任务分类和追踪
- **任务监控**：任务状态查询

## 任务状态

- `PENDING`: 等待执行
- `RUNNING`: 正在执行
- `COMPLETED`: 执行完成
- `FAILED`: 执行失败
- `CANCELLED`: 已取消
"""
taolib.testing.task_queue.server.api.tasks.get_task_service(request: fastapi.Request)#

获取任务服务。

class taolib.testing.task_queue.server.api.tasks.TaskListResponse#

Bases: pydantic.BaseModel

任务列表响应。

items: list[taolib.testing.task_queue.models.TaskResponse]#
total: int#
class taolib.testing.task_queue.server.api.tasks.SubmitTaskRequest#

Bases: pydantic.BaseModel

提交任务请求。

task_type: str#
params: dict#
priority: taolib.testing.task_queue.models.TaskPriority#
max_retries: int#
retry_delays: list[int]#
idempotency_key: str | None#
tags: list[str]#
class taolib.testing.task_queue.server.api.tasks.TaskActionResponse#

Bases: pydantic.BaseModel

任务操作响应。

task_id: str#
status: str#
message: str#
async taolib.testing.task_queue.server.api.tasks.list_tasks(request: fastapi.Request, skip: int = 0, limit: int = 20, status: taolib.testing.task_queue.models.TaskStatus | None = None, task_type: str | None = None, priority: taolib.testing.task_queue.models.TaskPriority | None = None) → TaskListResponse#

列出任务。

async taolib.testing.task_queue.server.api.tasks.get_task(request: fastapi.Request, task_id: str) → taolib.testing.task_queue.models.TaskResponse#

获取任务详情。

async taolib.testing.task_queue.server.api.tasks.submit_task(request: fastapi.Request, data: SubmitTaskRequest) → taolib.testing.task_queue.models.TaskResponse#

提交新任务。

async taolib.testing.task_queue.server.api.tasks.retry_task(request: fastapi.Request, task_id: str) → taolib.testing.task_queue.models.TaskResponse#

手动重试失败任务。

async taolib.testing.task_queue.server.api.tasks.cancel_task(request: fastapi.Request, task_id: str) → taolib.testing.task_queue.models.TaskResponse#

取消任务。

async taolib.testing.task_queue.server.api.tasks.delete_task(request: fastapi.Request, task_id: str) → None#

删除终态任务（仅 COMPLETED/FAILED/CANCELLED）。