🔍 Code Extractor

Marks a task as failed by updating its status, recording the error message, and setting the completion timestamp in a thread-safe manner.

Purpose

This function is used in a task management system to handle task failures. It updates the task's metadata in the active_tasks dictionary to reflect a failed state, stores the error information for debugging, and records when the failure occurred. The function uses a lock to ensure thread-safe access to the shared task dictionary, making it suitable for concurrent environments.

Source Code

def fail_task(task_id, error):
    """Mark task as failed"""
    with task_lock:
        if task_id in active_tasks:
            active_tasks[task_id]['status'] = 'failed'
            active_tasks[task_id]['error'] = str(error)
            active_tasks[task_id]['completed_at'] = datetime.now()

Parameters

Parameter Details

task_id: Unique identifier for the task to be marked as failed. Expected to be a key that exists in the active_tasks dictionary. Typically a string (UUID) or integer that was assigned when the task was created.

error: The error object or message that caused the task to fail. Can be an Exception object, string, or any object with a string representation. Will be converted to string using str() before storage.

Return Value

This function does not return any value (implicitly returns None). It performs side effects by modifying the active_tasks dictionary in place.

Dependencies

• datetime
• threading

Required Imports

from datetime import datetime
from threading import Lock

Usage Example

from datetime import datetime
from threading import Lock

# Setup required global variables
task_lock = Lock()
active_tasks = {
    'task-123': {
        'status': 'running',
        'created_at': datetime.now(),
        'error': None,
        'completed_at': None
    }
}

# Define the function
def fail_task(task_id, error):
    with task_lock:
        if task_id in active_tasks:
            active_tasks[task_id]['status'] = 'failed'
            active_tasks[task_id]['error'] = str(error)
            active_tasks[task_id]['completed_at'] = datetime.now()

# Usage example
try:
    # Some operation that might fail
    raise ValueError('Database connection failed')
except Exception as e:
    fail_task('task-123', e)

# Check the result
print(active_tasks['task-123']['status'])  # Output: 'failed'
print(active_tasks['task-123']['error'])   # Output: 'Database connection failed'

Best Practices

• Always ensure task_lock and active_tasks are properly initialized before calling this function
• The function silently does nothing if the task_id doesn't exist in active_tasks - consider adding logging or error handling if this is unexpected behavior
• The error parameter is converted to string, so stack traces from exceptions will be lost unless you use traceback.format_exc() before passing
• This function is thread-safe due to the lock, but ensure all other functions that access active_tasks also use the same lock
• Consider implementing a cleanup mechanism for old failed tasks to prevent memory leaks in long-running applications
• The function assumes the task dictionary already has 'status', 'error', and 'completed_at' keys - ensure tasks are properly initialized when created

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function fail_task 98.5% similar

  Marks a task as failed by updating its status, recording the error message, and setting the completion timestamp in a thread-safe manner.

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

• function complete_task_v1 79.0% 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 77.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 update_task_progress 67.0% 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_v1 63.8% 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