tasks.py•1.7 kB
"""
API routes for task management.
"""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Dict, Any, List, Optional
from src.database.database import get_db
from src.database import crud
router = APIRouter(
prefix="/tasks",
tags=["tasks"],
)
@router.get("/", response_model=List[Dict[str, Any]])
async def get_tasks(db: AsyncSession = Depends(get_db)):
"""Get all tasks."""
# This is a simplified implementation - in a real system, you'd want pagination
result = await db.execute("SELECT * FROM tasks ORDER BY created_at DESC LIMIT 100")
tasks = result.fetchall()
return [
{
"task_id": task.task_id,
"task_type": task.task_type,
"status": task.status,
"details": task.details,
"result": task.result,
"error": task.error,
"created_at": task.created_at,
"updated_at": task.updated_at
}
for task in tasks
]
@router.get("/{task_id}", response_model=Dict[str, Any])
async def get_task(task_id: str, db: AsyncSession = Depends(get_db)):
"""Get a task by ID."""
task = await crud.get_task(db, task_id)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task with ID {task_id} not found"
)
return {
"task_id": task.task_id,
"task_type": task.task_type,
"status": task.status,
"details": task.details,
"result": task.result,
"error": task.error,
"created_at": task.created_at,
"updated_at": task.updated_at
}