Sebastian Krüger
99c24adfe8
FastAPI async wrapper for Freepik cloud AI API supporting image generation (Mystic, Flux Dev/Pro, SeedReam), video generation (Kling, MiniMax, Seedance), image editing (upscale, relight, style transfer, expand, inpaint), and utilities (background removal, classifier, audio isolation). Includes async task tracking with polling, Docker containerization, and Gitea CI/CD workflow. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
import logging
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from fastapi.responses import FileResponse
|
|
|
|
from app.schemas.common import TaskDetail, TaskListResponse, TaskStatus
|
|
from app.services import task_tracker
|
|
from app.services.file_manager import get_result_path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix='/api/v1/tasks', tags=['tasks'])
|
|
|
|
|
|
@router.get('', response_model=TaskListResponse)
|
|
async def list_tasks(
|
|
status: Optional[TaskStatus] = Query(None),
|
|
limit: int = Query(20, ge=1, le=100),
|
|
offset: int = Query(0, ge=0),
|
|
):
|
|
tasks, total = task_tracker.list_tasks(status=status, limit=limit, offset=offset)
|
|
return TaskListResponse(
|
|
tasks=[_to_detail(t) for t in tasks],
|
|
total=total,
|
|
)
|
|
|
|
|
|
@router.get('/{task_id}', response_model=TaskDetail)
|
|
async def get_task(task_id: str):
|
|
task = task_tracker.get_task(task_id)
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail='Task not found')
|
|
return _to_detail(task)
|
|
|
|
|
|
@router.get('/{task_id}/result')
|
|
async def get_task_result(task_id: str):
|
|
task = task_tracker.get_task(task_id)
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail='Task not found')
|
|
if task['status'] != TaskStatus.completed:
|
|
raise HTTPException(status_code=409, detail=f'Task status is {task["status"].value}')
|
|
|
|
path = task.get('local_path') or get_result_path(task_id)
|
|
if not path:
|
|
# Fall back to redirect if we have a remote URL
|
|
if task.get('result_url'):
|
|
from fastapi.responses import RedirectResponse
|
|
return RedirectResponse(url=task['result_url'])
|
|
raise HTTPException(status_code=404, detail='Result file not found')
|
|
|
|
return FileResponse(path)
|
|
|
|
|
|
@router.delete('/{task_id}')
|
|
async def delete_task(task_id: str):
|
|
if not task_tracker.delete_task(task_id):
|
|
raise HTTPException(status_code=404, detail='Task not found')
|
|
return {'status': 'deleted', 'task_id': task_id}
|
|
|
|
|
|
def _to_detail(task: dict) -> TaskDetail:
|
|
result_url = None
|
|
if task['status'] == TaskStatus.completed:
|
|
result_url = f'/api/v1/tasks/{task["task_id"]}/result'
|
|
return TaskDetail(
|
|
task_id=task['task_id'],
|
|
status=task['status'],
|
|
created_at=task['created_at'],
|
|
updated_at=task['updated_at'],
|
|
progress=task.get('progress'),
|
|
result_url=result_url,
|
|
error=task.get('error'),
|
|
)
|