🔍 Code Extractor

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

Purpose

This function is designed for tracking progress of asynchronous or long-running tasks in a multi-threaded environment. It safely updates task progress information in a shared data structure (active_tasks dictionary) using a lock mechanism to prevent race conditions. Typically used in web applications or background job processing systems where multiple threads may be updating task statuses concurrently.

Source Code

def update_task_progress(task_id, progress):
    """Update task progress"""
    with task_lock:
        if task_id in active_tasks:
            active_tasks[task_id]['progress'] = progress
            active_tasks[task_id]['updated_at'] = datetime.now()

Parameters

Parameter Details

task_id: Unique identifier for the task whose progress needs to be updated. Expected to be a string or UUID that exists as a key in the active_tasks dictionary. If the task_id doesn't exist in active_tasks, the function silently does nothing.

progress: The progress value to set for the task. Can be any type (typically int, float, or string) representing the current state or completion percentage of the task. Common patterns include: integer percentage (0-100), float (0.0-1.0), or status string ('processing', 'completed', etc.).

Return Value

This function returns None (implicit). It performs an in-place update of the active_tasks dictionary and does not return any value. Side effects include modifying the 'progress' and 'updated_at' fields of the task entry if the task_id exists.

Dependencies

• datetime
• threading

Required Imports

from datetime import datetime
from threading import Lock

Usage Example

from datetime import datetime
from threading import Lock

# Required global setup
task_lock = Lock()
active_tasks = {}

# Initialize a task
task_id = 'task-123'
active_tasks[task_id] = {
    'progress': 0,
    'updated_at': datetime.now(),
    'status': 'started'
}

# Update task progress
update_task_progress(task_id, 25)
print(active_tasks[task_id]['progress'])  # Output: 25

# Update with different progress types
update_task_progress(task_id, 50.5)  # Float percentage
update_task_progress(task_id, 'processing')  # Status string
update_task_progress(task_id, 100)  # Completion

# Attempting to update non-existent task (safe, does nothing)
update_task_progress('non-existent-id', 50)

Best Practices

• Always ensure task_lock and active_tasks are properly initialized as global variables before calling this function
• Use consistent task_id format across your application (e.g., UUID strings) to avoid collisions
• Consider validating progress values before passing them to ensure they meet expected constraints (e.g., 0-100 for percentages)
• This function silently ignores updates for non-existent task_ids; consider adding logging or error handling if you need to track such cases
• The function is thread-safe due to the lock, but be aware that holding locks can impact performance in high-concurrency scenarios
• Consider implementing a cleanup mechanism to remove completed tasks from active_tasks to prevent memory leaks in long-running applications
• The updated_at timestamp uses datetime.now() which is timezone-naive; consider using datetime.now(timezone.utc) for timezone-aware timestamps
• If using this in a distributed system, note that this implementation only works within a single process; consider using Redis or a database for cross-process task tracking

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function complete_task 78.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 get_task_status_v1 78.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

• function complete_task_v1 75.4% 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 create_task_v1 69.8% similar

  Creates and registers a new background task entry in a thread-safe manner, initializing its status, progress, user information, and timestamps.

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

• function create_task 68.7% 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