🔍 Code Extractor

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

Purpose

This endpoint allows authenticated users to query the status of their background tasks by task ID. It enforces user ownership validation, ensuring users can only access their own tasks. The endpoint returns different response structures based on task status (pending, completed, or failed), including progress updates, results, or error messages as appropriate.

Source Code

def api_task_status(task_id):
    """Get status of a background task"""
    try:
        task = get_task_status(task_id)
        if not task:
            return jsonify({'error': 'Task not found'}), 404
        
        # Check if task belongs to current user
        current_user_email = session['user'].get('email', 'unknown')
        if task.get('user') != current_user_email:
            return jsonify({'error': 'Access denied'}), 403
        
        response_data = {
            'task_id': task_id,
            'status': task['status'],
            'progress': task.get('progress', ''),
            'created_at': task['created_at'].isoformat(),
        }
        
        if 'updated_at' in task:
            response_data['updated_at'] = task['updated_at'].isoformat()
        
        if task['status'] == 'completed':
            response_data['result'] = task['result']
            if 'completed_at' in task:
                response_data['completed_at'] = task['completed_at'].isoformat()
        elif task['status'] == 'failed':
            response_data['error'] = task['error']
            if 'completed_at' in task:
                response_data['completed_at'] = task['completed_at'].isoformat()
        
        return jsonify(response_data)
        
    except Exception as e:
        logger.error(f"Task status API error: {e}")
        return jsonify({'error': 'Failed to get task status'}), 500

Parameters

Parameter Details

task_id: String identifier for the background task. This is typically a UUID or unique string generated when the task was created. Used to look up the task status in the task tracking system.

Return Value

Returns a Flask JSON response tuple. On success (200): JSON object containing task_id, status, progress, created_at, and conditionally updated_at, completed_at, result (for completed tasks), or error (for failed tasks). On task not found (404): JSON with 'error' key. On access denied (403): JSON with 'error' key when task doesn't belong to current user. On server error (500): JSON with 'error' key.

Dependencies

• flask
• logging

Required Imports

from flask import jsonify
from flask import session
import logging

Usage Example

# Assuming Flask app setup with authentication and task management
# Client-side request example:
import requests

# After creating a background task and receiving task_id
task_id = 'abc-123-def-456'

# Query task status (with authentication cookies/headers)
response = requests.get(
    f'http://localhost:5000/api/task-status/{task_id}',
    cookies={'session': 'your_session_cookie'}
)

if response.status_code == 200:
    data = response.json()
    print(f"Status: {data['status']}")
    print(f"Progress: {data.get('progress', 'N/A')}")
    if data['status'] == 'completed':
        print(f"Result: {data['result']}")
    elif data['status'] == 'failed':
        print(f"Error: {data['error']}")
elif response.status_code == 404:
    print('Task not found')
elif response.status_code == 403:
    print('Access denied - not your task')

Best Practices

• Always check task ownership before returning task information to prevent unauthorized access
• Use proper HTTP status codes (404 for not found, 403 for forbidden, 500 for server errors)
• Include timestamps in ISO format for consistent datetime representation across clients
• Log errors with sufficient context for debugging while returning generic error messages to clients
• Conditionally include fields in response based on task status to avoid exposing irrelevant data
• Ensure the get_task_status function returns consistent task object structure with all required fields
• Consider implementing rate limiting to prevent status polling abuse
• The session must contain a 'user' dictionary with an 'email' key for authorization to work properly

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function get_task_status 85.2% similar

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

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

• function api_task_status 85.0% 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_cancel_task 79.9% similar

  Flask API endpoint that cancels a background task if it belongs to the authenticated user and is currently processing.

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

• function api_chat_status 73.1% 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 api_index_progress 67.9% 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