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

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

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:

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:

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:

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): Thin FastAPI endpoints that delegate immediately, enabling non-blocking HTTP responses
• Manager layer (api/tasks/manager.py): In-memory asyncio orchestration with execute_task, state tracking, progress updates, and automated cleanup
• Model layer (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.

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