# How the `tasks` Router Manages Asynchronous Operations in Pixelle-Video

> Learn how the Pixelle-Video tasks router manages asynchronous operations using asyncio TaskManager for efficient, non-blocking HTTP requests and background task orchestration.

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

---

**The `tasks` router in Pixelle-Video delegates all long-running work to a `TaskManager` that orchestrates `asyncio` tasks, tracks progress in memory, and handles lifecycle cleanup—while the router itself remains a thin, non-blocking HTTP layer.**

The `tasks` router is a critical component in the [AIDC-AI/Pixelle-Video](https://github.com/AIDC-AI/Pixelle-Video) repository, enabling the API to handle video generation as asynchronous background jobs. Rather than blocking HTTP requests during GPU-intensive inference, the router manages asynchronous operations through a clean separation between HTTP handling and task execution.

## Router Architecture: Thin HTTP Layer Over Async Manager

The `tasks` router in [`api/routers/tasks.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/routers/tasks.py) follows FastAPI best practices by remaining stateless and delegation-heavy. Each endpoint is declared as `async def` but performs no heavy computation—instead, it accesses the `TaskManager`'s in-memory structures.

### Endpoint-to-Manager Mapping

| Router Endpoint | Manager Method | Purpose |
|-----------------|--------------|---------|
| `GET /tasks` | `task_manager.list_tasks` | Filtered list with status and pagination |
| `GET /tasks/{task_id}` | `task_manager.get_task` | Single task details, progress, and result |
| `DELETE /tasks/{task_id}` | `task_manager.cancel_task` | Cancel pending/running tasks |

The router's role in managing asynchronous operations is strictly coordinative—creating task entries, returning immediate responses with `task_id`, and enabling client polling.

## TaskManager: Core Async Execution Engine

The `TaskManager` in [`api/tasks/manager.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py) implements the actual asynchronous orchestration pattern. It maintains two critical data structures:

- **`_tasks`**: Dictionary mapping `task_id` to `Task` objects (state, progress, results)
- **`_task_futures`**: Dictionary mapping `task_id` to `asyncio.Task` futures for cancellation

### The `execute_task` Method

The `execute_task` method is the heart of the manager's async operation handling:

```python
async def execute_task(self, task_id: str, coro_func: Callable, *args, **kwargs):
    task = self._tasks.get(task_id)
    # ... validation ...

    
    async def _execute():
        try:
            task.status = TaskStatus.RUNNING
            task.started_at = datetime.now()
            result = await coro_func(*args, **kwargs)  # User's coroutine

            task.status = TaskStatus.COMPLETED
            task.result = result
            task.completed_at = datetime.now()
        except Exception as e:
            task.status = TaskStatus.FAILED
            task.error = str(e)
            task.completed_at = datetime.now()
    
    future = asyncio.create_task(_execute())
    self._task_futures[task_id] = future

```

Key characteristics of this pattern:
- **Immediate return**: The router receives `task_id` before work begins
- **Background execution**: `asyncio.create_task` schedules the coroutine without blocking
- **State tracking**: The wrapper updates `TaskStatus` through the full lifecycle
- **Exception isolation**: Failures in user code are caught and stored, not propagated

## Progress Tracking and Lifecycle Management

The `tasks` router's approach to managing asynchronous operations includes comprehensive progress reporting and automatic cleanup.

### Progress Updates

The `TaskManager.update_progress` method allows any running coroutine to report incremental progress:

```python
def update_progress(self, task_id: str, current: int, total: int, message: str = ""):
    task = self._tasks.get(task_id)
    if task:
        task.progress = TaskProgress(
            current=current,
            total=total,
            percentage=(current / total * 100) if total > 0 else 0,
            message=message
        )

```

This enables real-time polling via `GET /tasks/{task_id}` without WebSocket complexity.

### Automated Cleanup

The manager implements a background cleanup loop to prevent memory leaks:

```python
async def _cleanup_loop(self):
    while self._running:
        await asyncio.sleep(self.config.task_cleanup_interval)
        self._cleanup_old_tasks()

def _cleanup_old_tasks(self):
    cutoff = datetime.now() - timedelta(seconds=self.config.task_retention_time)
    for task_id, task in list(self._tasks.items()):
        if task.completed_at and task.completed_at < cutoff:
            future = self._task_futures.pop(task_id, None)
            if future and not future.done():
                future.cancel()
            del self._tasks[task_id]

```

Configuration via `api_config`:
- `task_cleanup_interval`: How often to run cleanup (seconds)
- `task_retention_time`: How long to retain completed tasks (seconds)

## Complete Implementation Example

Here's how to create, execute, and poll an asynchronous video generation task using the `tasks` router infrastructure:

```python
from api.tasks import task_manager, TaskType
import asyncio
import uuid

# 1. Create task entry (returns immediately)

task = task_manager.create_task(
    task_type=TaskType.VIDEO_GENERATION,
    request_params={"prompt": "A futuristic city skyline", "duration": 5.0}
)

# 2. Define the actual async work

async def generate_video(prompt: str, duration: float) -> str:
    total_steps = 50
    for step in range(total_steps):
        # Simulate inference step

        await asyncio.sleep(0.1)
        # Report progress

        task_manager.update_progress(
            task.task_id, 
            current=step + 1, 
            total=total_steps,
            message=f"Generating frame group {step+1}/{total_steps}"
        )
    return f"https://cdn.example.com/videos/{uuid.uuid4()}.mp4"

# 3. Execute asynchronously (non-blocking)

await task_manager.execute_task(
    task_id=task.task_id,
    coro_func=generate_video,
    **task.request_params
)

# 4. Client polls via HTTP: GET /tasks/{task_id}

# Response includes: status, progress percentage, message, result or error

```

## Summary

The `tasks` router in Pixelle-Video manages asynchronous operations through a clean architectural separation:

- **Router layer** ([`api/routers/tasks.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/routers/tasks.py)): Thin FastAPI endpoints that delegate immediately, enabling non-blocking HTTP responses
- **Manager layer** ([`api/tasks/manager.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py)): In-memory `asyncio` orchestration with `execute_task`, state tracking, progress updates, and automated cleanup
- **Model layer** ([`api/tasks/models.py`](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/models.py)): Structured `Task`, `TaskStatus`, and `TaskProgress` schemas for API consistency

Key implementation patterns include using `asyncio.create_task` for background execution, wrapping user coroutines for exception isolation, and implementing a configurable cleanup loop to prevent memory leaks.

## Frequently Asked Questions

### What happens if a task fails during execution?

The `execute_task` wrapper in `TaskManager` catches all exceptions, sets `task.status` to `TaskStatus.FAILED`, stores the error message in `task.error`, and records `task.completed_at`. The HTTP endpoint `GET /tasks/{task_id}` returns this error information to the client without crashing the server.

### How does the router handle task cancellation?

When a client sends `DELETE /tasks/{task_id}`, the router invokes `task_manager.cancel_task`. This method retrieves the `asyncio.Task` future from `_task_futures`, calls `future.cancel()` if still pending, and updates the task status to `cancelled`. Cancellation only works for tasks not yet completed or failed.

### Why use in-memory storage instead of a persistent database?

The `TaskManager` stores tasks in `_tasks` dictionary for low-latency access and simplified deployment. This design suits short-lived video generation jobs where clients poll actively during execution. The automated `_cleanup_old_tasks` removes stale entries, preventing unbounded memory growth. For production persistence, the repository could extend `TaskManager` with Redis or database backends while keeping the same async interface.