🔍 Code Extractor

Removes tasks from the active_tasks dictionary that are older than 1 hour (3600 seconds) based on their creation timestamp, using thread-safe locking.

Purpose

This function performs periodic cleanup of stale tasks to prevent memory leaks and maintain system performance. It iterates through all active tasks, calculates their age, and removes those exceeding the 1-hour threshold. The function is designed to be called periodically (e.g., by a scheduler or background thread) to maintain a clean task registry in a multi-threaded Flask application.

Source Code

def cleanup_old_tasks():
    """Clean up tasks older than 1 hour"""
    with task_lock:
        current_time = datetime.now()
        to_remove = []
        for task_id, task in active_tasks.items():
            age = current_time - task['created_at']
            if age.total_seconds() > 3600:  # 1 hour
                to_remove.append(task_id)
        for task_id in to_remove:
            del active_tasks[task_id]
            logger.info(f"Cleaned up old task: {task_id}")

Return Value

This function returns None (implicitly). It performs side effects by modifying the global active_tasks dictionary and logging cleanup operations.

Dependencies

• datetime
• threading
• logging

Required Imports

from datetime import datetime
from threading import Lock
import logging

Usage Example

import logging
from datetime import datetime
from threading import Lock

# Setup required globals
task_lock = Lock()
active_tasks = {
    'task_1': {'created_at': datetime.now(), 'status': 'running'},
    'task_2': {'created_at': datetime(2023, 1, 1), 'status': 'completed'}
}
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Define the function
def cleanup_old_tasks():
    with task_lock:
        current_time = datetime.now()
        to_remove = []
        for task_id, task in active_tasks.items():
            age = current_time - task['created_at']
            if age.total_seconds() > 3600:
                to_remove.append(task_id)
        for task_id in to_remove:
            del active_tasks[task_id]
            logger.info(f"Cleaned up old task: {task_id}")

# Call the cleanup function
cleanup_old_tasks()
print(f"Remaining tasks: {len(active_tasks)}")

Best Practices

• Ensure task_lock is properly initialized as a threading.Lock() before calling this function
• Ensure active_tasks dictionary exists and contains tasks with 'created_at' datetime fields
• Call this function periodically using a scheduler (e.g., APScheduler) or background thread to prevent memory buildup
• The 1-hour threshold (3600 seconds) is hardcoded; consider making it configurable for different use cases
• Ensure logger is properly configured before calling this function to capture cleanup events
• This function modifies global state; ensure proper synchronization if multiple cleanup threads might run
• Consider adding error handling for malformed task entries that might not have 'created_at' fields
• The function uses a two-pass approach (collect then delete) to avoid modifying dictionary during iteration

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function cleanup_old_tasks 93.6% similar

  Removes tasks from the active_tasks dictionary that are older than 1 hour (3600 seconds) based on their creation timestamp.

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

• function cleanup_old_documents 66.0% similar

  Periodically removes documents and their associated files that are older than 2 hours from the uploaded_documents dictionary, cleaning up both file system storage and memory.

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

• function create_task_v1 62.7% 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 62.6% 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 complete_task_v1 58.1% 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