🔍 Code Extractor

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.

Purpose

This function is used to initialize and track background tasks in a Flask web application. It creates a task entry with initial metadata including status, progress, user association, creation timestamp, and placeholders for results and errors. The function uses a lock to ensure thread-safe access to the shared active_tasks dictionary, making it suitable for concurrent task management in a multi-threaded environment.

Source Code

def create_task(task_id, user_id):
    """Create a new background task"""
    with task_lock:
        active_tasks[task_id] = {
            'status': 'processing',
            'progress': 'Starting...',
            'user': user_id,
            'created_at': datetime.now(),
            'result': None,
            'error': None
        }

Parameters

Parameter Details

task_id: A unique identifier for the task being created. This should be a string or UUID that uniquely identifies the background task. It serves as the key in the active_tasks dictionary.

user_id: The identifier of the user who initiated the task. This is used to associate the task with a specific user for tracking, authorization, and filtering purposes.

Return Value

This function does not return any value (returns None implicitly). It performs a side effect by modifying the global active_tasks dictionary to add a new task entry.

Dependencies

• datetime
• threading

Required Imports

from datetime import datetime
from threading import Lock

Usage Example

from datetime import datetime
from threading import Lock
import uuid

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

# Create a new task
task_id = str(uuid.uuid4())
user_id = 'user123'
create_task(task_id, user_id)

# Verify task was created
with task_lock:
    task_info = active_tasks.get(task_id)
    print(f"Task status: {task_info['status']}")
    print(f"Task progress: {task_info['progress']}")
    print(f"Task user: {task_info['user']}")
    print(f"Created at: {task_info['created_at']}")

Best Practices

• Always ensure task_lock and active_tasks are properly initialized as global variables before calling this function
• Use unique task_id values (e.g., UUID) to prevent task collisions
• This function should be called at the start of a background task execution flow
• Consider implementing task cleanup mechanisms to prevent memory leaks from accumulating tasks in active_tasks
• The task_lock should be the same Lock instance used by all functions that access active_tasks to ensure proper synchronization
• After creating a task, implement proper error handling and status updates in the actual task execution code
• Consider adding task expiration or cleanup logic to remove old completed tasks from active_tasks

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function create_task_v1 91.3% 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 get_task_status_v1 70.2% 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 update_task_progress 68.7% 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 fail_task 65.0% 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 64.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