🔍 Code Extractor

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

Purpose

This function is used to initialize a background task tracking system, typically for long-running operations like document processing or API calls. It creates a task entry in a shared dictionary (active_tasks) with initial metadata including status, progress message, user email, creation timestamp, and placeholders for results and errors. The function uses a lock to ensure thread-safe access to the shared task dictionary, making it suitable for concurrent web applications.

Source Code

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

Parameters

Parameter Details

task_id: A unique identifier for the task, typically a UUID string or similar unique value. This ID is used as the key to store and retrieve the task from the active_tasks dictionary.

user_email: The email address of the user who initiated the task. Used for tracking task ownership and potentially for notifications or access control.

Return Value

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

Dependencies

• threading
• datetime

Required Imports

from threading import Lock
from datetime import datetime

Usage Example

from threading import Lock
from datetime import datetime
import uuid

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

# Define the function
def create_task(task_id, user_email):
    with task_lock:
        active_tasks[task_id] = {
            'status': 'processing',
            'progress': 'Starting request...',
            'user': user_email,
            'created_at': datetime.now(),
            'result': None,
            'error': None
        }

# Usage example
task_id = str(uuid.uuid4())
user_email = 'user@example.com'
create_task(task_id, user_email)

# Verify task was created
print(f"Task created: {active_tasks[task_id]['status']}")
print(f"User: {active_tasks[task_id]['user']}")

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
• Consider implementing a cleanup mechanism to remove completed tasks from active_tasks to prevent memory leaks
• Validate user_email format before passing to this function for data integrity
• This function should be paired with complementary functions to update task status, retrieve task information, and clean up completed tasks
• In production environments, consider using a more robust task queue system like Celery or Redis for distributed systems
• The created_at timestamp uses datetime.now() which is timezone-naive; consider using datetime.now(timezone.utc) for timezone-aware timestamps

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

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

• function update_task_progress 69.8% 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 complete_task_v1 67.3% 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 66.3% 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 64.9% 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