# How Pixelle-Video Implements Asynchronous Task Management for Video Generation

> Discover how Pixelle-Video implements asynchronous task management with an in-memory system, Pydantic Task model, and FastAPI for efficient video generation.

- Repository: [AIDC-AI/Pixelle-Video](https://github.com/AIDC-AI/Pixelle-Video)
- Tags: internals
- Published: 2026-04-23

---

**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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/models.py)

Every async operation in Pixelle-Video is represented by a **`Task`** object defined in [`api/tasks/models.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/models.py#L45-L65)

## The TaskManager Singleton in [`api/tasks/manager.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py)

The **`TaskManager`** class in [`api/tasks/manager.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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:

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

```

Source: [`api/tasks/manager.py#L78-L94`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py#L78-L94)

### Background Execution

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

```python
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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py#L105-L138)

### Progress Tracking

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

```python
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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py#L34-L42), [`api/tasks/manager.py#L45-L62`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/routers/video.py) initiates background video generation:

```python
@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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/routers/video.py#L78-L82)

### Task Management Endpoints

The dedicated task router in [`api/routers/tasks.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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:

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

```

Cancelling a task:

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

```

Source: [`api/routers/tasks.py#L28-L76`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/routers/tasks.py#L28-L76)

## Complete Task Lifecycle Example

Submitting an async video generation job:

```bash
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:

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

```

Polling while running:

```json
{
  "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:

```json
{
  "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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/routers/video.py), [`api/routers/tasks.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/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.