🔍 Code Extractor

Flask API endpoint that retrieves the current status of a background task by its task ID from an in-memory active_tasks dictionary.

Purpose

This endpoint allows clients to poll for the status of long-running background tasks (likely document processing or indexing operations). It returns the task's current state from the active_tasks global dictionary, enabling asynchronous operation monitoring. Returns a 404 error if the task ID is not found.

Source Code

def get_task_status(task_id):
    """Get status of background task"""
    if task_id not in active_tasks:
        return jsonify({'error': 'Task not found'}), 404
    
    return jsonify(active_tasks[task_id])

Parameters

Parameter Details

task_id: String identifier for the background task. This is extracted from the URL path parameter (/api/task/<task_id>). Expected to be a unique identifier (likely a UUID) that was returned when the task was initially created. Must exist as a key in the active_tasks dictionary.

Return Value

Returns a Flask JSON response tuple. On success: (jsonify(task_dict), 200) where task_dict contains the task status information from active_tasks[task_id] (structure depends on how tasks are stored, typically includes fields like 'status', 'progress', 'result', etc.). On failure: (jsonify({'error': 'Task not found'}), 404) when task_id doesn't exist in active_tasks.

Dependencies

• flask
• flask_login

Required Imports

from flask import Blueprint
from flask import jsonify
from flask_login import login_required

Usage Example

# Assuming Flask app setup with authentication
# Client-side usage (JavaScript fetch example):

# After starting a background task that returns task_id:
const taskId = 'abc-123-def-456';

// Poll for task status
fetch(`/api/task/${taskId}`, {
  method: 'GET',
  credentials: 'include',  // Include session cookie
  headers: {
    'Content-Type': 'application/json'
  }
})
.then(response => response.json())
.then(data => {
  if (data.error) {
    console.error('Task not found:', data.error);
  } else {
    console.log('Task status:', data.status);
    console.log('Task progress:', data.progress);
  }
})
.catch(error => console.error('Error:', error));

# Server-side context (how active_tasks might be structured):
# active_tasks = {
#   'task-uuid-1': {'status': 'processing', 'progress': 50, 'message': 'Indexing documents'},
#   'task-uuid-2': {'status': 'completed', 'progress': 100, 'result': 'Success'}
# }

Best Practices

• This endpoint is designed for polling; consider implementing WebSocket or Server-Sent Events for real-time updates in production
• The active_tasks dictionary should be thread-safe if tasks are created/updated from multiple threads (consider using threading.Lock)
• Implement task cleanup to prevent memory leaks - remove completed tasks from active_tasks after a certain time
• Consider adding pagination or filtering if many tasks exist
• Add rate limiting to prevent excessive polling requests
• The task_id should be validated/sanitized if not already handled by Flask routing
• Consider adding authorization checks to ensure users can only access their own tasks
• For production, consider using a persistent store (Redis, database) instead of in-memory dictionary for task status

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function api_task_status 86.8% similar

  Flask API endpoint that retrieves and returns the status of asynchronous tasks (chat or indexing operations) by task ID.

  From: /tf/active/vicechatdev/docchat/app.py

• function api_task_status_v1 85.2% similar

  Flask API endpoint that retrieves and returns the status of a background task, with user authorization checks.

  From: /tf/active/vicechatdev/vice_ai/app.py

• function api_index_progress 75.5% similar

  Flask API endpoint that retrieves the current progress status of an asynchronous indexing task by its task ID.

  From: /tf/active/vicechatdev/docchat/app.py

• function api_chat_status 73.5% similar

  Flask API endpoint that retrieves the detailed status of a chat task by delegating to the api_task_status function using the provided task_id.

  From: /tf/active/vicechatdev/docchat/app.py

• function get_task_status_v1 73.1% similar

  Thread-safe function that retrieves the current status of a task from a shared dictionary using a task identifier.

  From: /tf/active/vicechatdev/docchat/app.py