How Pixelle-Video Implements Asynchronous Task Management for Video Generation

Pixelle-Video uses a lightweight in-memory task management system built around a Pydantic-based Task model, a singleton TaskManager class, and FastAPI routers that expose async operations through a clean REST API.

The AIDC-AI/Pixelle-Video repository implements a complete asynchronous task management system that handles background video generation without blocking client requests. This article examines the three core components—the data model, the manager class, and the API layer—and shows how they work together to create, execute, monitor, and cancel async tasks.

The Task Data Model in api/tasks/models.py

Every async operation in Pixelle-Video is represented by a Task object defined in api/tasks/models.py. This Pydantic data class stores all metadata required to track an asynchronous job from creation through completion.

The Task model includes:

• task_id: UUIDv4 primary identifier
• task_type: Enum distinguishing operation types (e.g., VIDEO_GENERATION)
• status: Lifecycle state (PENDING, RUNNING, COMPLETED, FAILED, CANCELLED)
• progress: Optional TaskProgress object with current, total, percentage, and message
• result: Arbitrary JSON payload returned on successful completion
• error: Error details captured on failure
• Timestamps: created_at, started_at, completed_at, cancelled_at

Source: api/tasks/models.py#L45-L65

The TaskManager Singleton in api/tasks/manager.py

The TaskManager class in api/tasks/manager.py serves as the central coordinator for all async operations. Implemented as a singleton, it maintains an in-memory dictionary of active tasks and provides the complete lifecycle API.

Task Creation

The create_task() method instantiates a new Task with status PENDING and stores it in the internal _tasks dictionary:

task = task_manager.create_task(
    task_type=TaskType.VIDEO_GENERATION,
    request_params=request_body.model_dump()
)

Source: api/tasks/manager.py#L78-L94

Background Execution

The execute_task() method runs the actual async work without blocking the caller:

await task_manager.execute_task(
    task_id=task.task_id,
    coro_func=execute_video_generation
)

The internal _execute wrapper handles state transitions:

1. Sets status to RUNNING and records started_at
2. Awaits the user-provided coroutine
3. On success: stores result, sets status to COMPLETED, records completed_at
4. On exception: captures error details, sets status to FAILED

Source: api/tasks/manager.py#L105-L138

Progress Tracking

Long-running operations can report progress via update_progress():

task_manager.update_progress(
    task_id=task_id,
    current=2,
    total=5,
    message="Generating scene 2/5"
)

The video generation endpoint includes a commented hook for future integration of fine-grained progress callbacks.

Source: api/tasks/manager.py#L81-L89

Task Cancellation

The cancel_task() method terminates pending or running tasks:

• Cancels the underlying asyncio.Task if still pending
• Sets status to CANCELLED
• Records cancelled_at timestamp

Source: api/tasks/manager.py#L129-L132

Automatic Cleanup

A background _cleanup_loop runs at configurable intervals (api_config.task_cleanup_interval) to prevent unbounded memory growth. It removes tasks older than api_config.task_retention_time with terminal status (COMPLETED, FAILED, or CANCELLED).

Source: api/tasks/manager.py#L34-L42, api/tasks/manager.py#L45-L62

FastAPI Routers for Task Management API

The task system exposes HTTP endpoints through two router modules that translate between client requests and TaskManager operations.

Video Generation Endpoint

The /video/generate/async endpoint in api/routers/video.py initiates background video generation:

@router.post("/generate/async")
async def generate_video_async(request_body: VideoGenerationRequest):
    task = task_manager.create_task(
        task_type=TaskType.VIDEO_GENERATION,
        request_params=request_body.model_dump()
    )
    
    async def execute_video_generation():
        # Heavy video generation work here

        result = await generate_video_from_request(request_body)
        return result
    
    await task_manager.execute_task(
        task_id=task.task_id,
        coro_func=execute_video_generation
    )
    
    return {"task_id": task.task_id}

Source: api/routers/video.py#L78-L82

Task Management Endpoints

The dedicated task router in api/routers/tasks.py provides full CRUD-style operations:

Endpoint	Method	Description
GET /tasks	list_tasks()	Returns all tasks with optional filtering
GET /tasks/{task_id}	get_task()	Retrieves specific task by ID
DELETE /tasks/{task_id}	cancel_task()	Cancels pending or running task

Polling task status:

curl http://localhost:8000/api/tasks/b3c1d9f2-6a44-4e8b-9c2d-7f5e1a0e9c3f

Cancelling a task:

curl -X DELETE http://localhost:8000/api/tasks/b3c1d9f2-6a44-4e8b-9c2d-7f5e1a0e9c3f

Source: api/routers/tasks.py#L28-L76

Complete Task Lifecycle Example

Submitting an async video generation job:

curl -X POST http://localhost:8000/api/video/generate/async \
     -H "Content-Type: application/json" \
     -d '{
           "text": "A short story about cats.",
           "frame_template": "default",
           "mode": "story",
           "title": "Cat Tale",
           "n_scenes": 3,
           "media_width": 1280,
           "media_height": 720
         }'

Response with task ID:

{
  "task_id": "b3c1d9f2-6a44-4e8b-9c2d-7f5e1a0e9c3f"
}

Polling while running:

{
  "task_id": "b3c1d9f2-6a44-4e8b-9c2d-7f5e1a0e9c3f",
  "task_type": "video_generation",
  "status": "running",
  "progress": {
    "current": 2,
    "total": 5,
    "percentage": 40.0,
    "message": "Generating scene 2/5"
  },
  "created_at": "2026-04-23T10:12:34.567890",
  "started_at": "2026-04-23T10:12:35.001234"
}

Final completed response:

{
  "task_id": "b3c1d9f2-6a44-4e8b-9c2d-7f5e1a0e9c3f",
  "status": "completed",
  "result": {
    "video_url": "http://localhost:8000/api/files/20260423_101235/final.mp4",
    "duration": 12.4,
    "file_size": 8423671
  },
  "completed_at": "2026-04-23T10:12:50.123456"
}

Summary

Pixelle-Video's asynchronous task management system delivers a production-ready solution for background video generation through three integrated layers:

• Task model (api/tasks/models.py) provides a Pydantic-based schema for task state, progress, and results with strict type safety
• TaskManager singleton (api/tasks/manager.py) handles the complete lifecycle: creation, async execution with proper state transitions, progress updates, cancellation, and automatic cleanup of finished tasks
• FastAPI routers (api/routers/video.py, api/routers/tasks.py) expose REST endpoints for submitting jobs, polling status, and managing active tasks

The in-memory architecture prioritizes simplicity and low latency while remaining extensible—replacing the dictionary store with Redis or a database would require changes only to TaskManager's internal storage methods, with no impact on the public API contract.

Frequently Asked Questions

How does Pixelle-Video handle task persistence across server restarts?

The current implementation stores tasks in an in-memory Python dictionary within the TaskManager singleton. This means tasks are lost on server restart. According to the source code in api/tasks/manager.py, the design intentionally prioritizes simplicity and low overhead for single-instance deployments. For production environments requiring persistence, the TaskManager class would need modification to replace the _tasks: Dict[str, Task] store with a Redis hash or database table, while preserving the same public method signatures.

What task statuses are available and how do they transition?

Pixelle-Video defines five task statuses in api/tasks/models.py: PENDING (initial state after creation), RUNNING (set when execution begins), COMPLETED (successful finish with result stored), FAILED (exception caught with error details), and CANCELLED ( explicit cancellation request). The TaskManager._execute wrapper in api/tasks/manager.py handles all transitions automatically—only PENDING → RUNNING → (COMPLETED | FAILED) and PENDING/RUNNING → CANCELLED are valid paths.

How can I implement progress updates for custom task types?

The TaskManager provides update_progress(task_id, current, total, message) for fine-grained progress reporting. According to api/tasks/manager.py#L81-L89, this method constructs a TaskProgress object and stores it in the task's progress field. For custom long-running tasks, pass a callback closure into your coroutine that invokes task_manager.update_progress() at appropriate milestones. The video generation endpoint in api/routers/video.py includes a commented placeholder demonstrating this pattern for future scene-level progress reporting.

What limits exist on concurrent task execution?

The current implementation in api/tasks/manager.py does not enforce explicit concurrency limits—each execute_task call creates an independent asyncio.Task that runs immediately. The practical limit depends on your Python event loop capacity and the resource intensity of underlying operations (video generation is typically CPU/GPU bound). For production deployments, you would extend TaskManager with a asyncio.Semaphore or integrate with a task queue like Celery or RQ to constrain concurrent video generation jobs and prevent resource exhaustion.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}