🔍 Code Extractor

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

Purpose

This function provides a thread-safe way to query the status of asynchronous tasks in a Flask application. It uses a lock mechanism to prevent race conditions when accessing the shared active_tasks dictionary. The function is typically used in task monitoring systems where multiple threads may be updating and reading task statuses concurrently.

Source Code

def get_task_status(task_id):
    """Get current task status"""
    with task_lock:
        return active_tasks.get(task_id, None)

Parameters

Parameter Details

task_id: A unique identifier (likely a string or UUID) used to look up the task in the active_tasks dictionary. This ID should correspond to a task that was previously registered in the active_tasks collection.

Return Value

Returns the task status object associated with the given task_id if it exists in the active_tasks dictionary, or None if the task_id is not found. The exact structure of the returned status object depends on how tasks are stored in the active_tasks dictionary, but typically includes information like task state, progress, results, or error messages.

Dependencies

• threading

Required Imports

from threading import Lock

Usage Example

from threading import Lock

# Required setup
task_lock = Lock()
active_tasks = {
    'task-123': {'status': 'running', 'progress': 50},
    'task-456': {'status': 'completed', 'result': 'success'}
}

# Function definition
def get_task_status(task_id):
    with task_lock:
        return active_tasks.get(task_id, None)

# Usage
status = get_task_status('task-123')
if status:
    print(f"Task status: {status['status']}")
else:
    print("Task not found")

# Returns None for non-existent tasks
missing = get_task_status('task-999')
print(missing)  # Output: None

Best Practices

• Ensure that task_lock is a threading.Lock object initialized at module level before calling this function
• Ensure that active_tasks is a dictionary initialized at module level before calling this function
• Always check if the returned value is None before attempting to access its properties
• Use consistent task_id formats throughout your application (e.g., always use UUIDs or always use strings)
• Consider implementing task cleanup mechanisms to prevent the active_tasks dictionary from growing indefinitely
• The lock acquisition is handled automatically by the 'with' statement, ensuring proper release even if exceptions occur
• This function is read-only and thread-safe, making it suitable for concurrent access from multiple threads or request handlers

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function update_task_progress 78.1% similar

  Thread-safe function that updates the progress and timestamp of an active task identified by task_id in a shared dictionary.

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

• function get_task_status 73.1% 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 complete_task_v1 72.6% similar

  Thread-safe function that marks a task as completed by updating its status, storing the result, and recording the completion timestamp in a shared task dictionary.

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

• function complete_task 70.9% similar

  Updates the status of a task to 'completed' in a thread-safe manner, storing the result and completion timestamp.

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

• function create_task 70.2% similar

  Creates and registers a new background task in a thread-safe manner by initializing its metadata in a shared dictionary with status tracking, progress information, and timestamps.

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