🔍 Code Extractor

Flask route handler that retrieves the progress status of a running analysis task and performs cleanup of completed/failed analyses after a timeout period.

Purpose

This endpoint provides real-time progress tracking for asynchronous analysis tasks. It returns the current status, progress percentage, and other metadata for a given analysis ID. Additionally, it implements automatic cleanup logic to remove stale progress data after 5 minutes for completed/failed analyses and triggers periodic cleanup of old sessions when auto-cleanup is enabled.

Source Code

def get_analysis_progress(analysis_id):
    """Get progress of running analysis"""
    if analysis_id not in analysis_progress:
        return jsonify({'error': 'Analysis not found'}), 404
    
    progress_data = analysis_progress[analysis_id].copy()
    
    # Clean up completed/failed analyses after some time
    if progress_data['status'] in ['completed', 'failed']:
        started_at = datetime.fromisoformat(progress_data['started_at'])
        if (datetime.now() - started_at).total_seconds() > 300:  # 5 minutes
            del analysis_progress[analysis_id]
            
            # Periodic cleanup trigger - run cleanup occasionally
            if app_config.AUTO_CLEANUP_ENABLED and len(analysis_progress) % 20 == 0:  # Every 20 completed analyses
                try:
                    analysis_service.cleanup_old_sessions(days_old=app_config.CLEANUP_OLD_SESSIONS_DAYS)
                except Exception as periodic_cleanup_error:
                    logger.warning(f"Periodic cleanup warning: {str(periodic_cleanup_error)}")
    
    return jsonify(progress_data)

Parameters

Parameter Details

analysis_id: String identifier for the analysis task. This ID is used as a key to look up progress information in the analysis_progress dictionary. Expected to be a UUID or unique string generated when the analysis was initiated.

Return Value

Returns a Flask JSON response. On success (200): returns a dictionary containing progress data with keys like 'status' (running/completed/failed), 'progress' (percentage), 'started_at' (ISO format datetime string), and potentially 'result' or 'error' fields. On failure (404): returns {'error': 'Analysis not found'} when the analysis_id doesn't exist in the progress tracking dictionary.

Dependencies

• flask
• datetime
• logging
• pathlib
• shutil
• threading

Required Imports

from flask import Flask, jsonify
from datetime import datetime
import logging

Conditional/Optional Imports

These imports are only needed under specific conditions:

from services import StatisticalAnalysisService

from config import Config

Usage Example

# Assuming Flask app setup and global variables exist
# analysis_progress = {}
# app = Flask(__name__)
# app_config = Config()
# analysis_service = StatisticalAnalysisService()

# Client-side usage (HTTP request):
import requests

analysis_id = 'abc-123-def-456'
response = requests.get(f'http://localhost:5000/analysis_progress/{analysis_id}')

if response.status_code == 200:
    progress = response.json()
    print(f"Status: {progress['status']}")
    print(f"Progress: {progress.get('progress', 0)}%")
else:
    print(f"Error: {response.json()['error']}")

# Server-side: Progress data structure example
# analysis_progress[analysis_id] = {
#     'status': 'running',
#     'progress': 45,
#     'started_at': datetime.now().isoformat(),
#     'message': 'Processing data...'
# }

Best Practices

• This endpoint is designed for polling; clients should implement reasonable polling intervals (e.g., every 1-2 seconds) to avoid overwhelming the server
• The 5-minute cleanup timeout for completed/failed analyses means clients should retrieve results within this window
• The periodic cleanup (every 20 completed analyses) may cause brief delays; consider running cleanup in a background thread for production use
• Ensure proper error handling for the cleanup_old_sessions() call as it's wrapped in try-except but only logs warnings
• The analysis_progress dictionary should be thread-safe if multiple workers are used; consider using threading.Lock or a proper cache backend (Redis, Memcached) for production
• The function modifies the global analysis_progress dictionary by deleting entries, which could cause race conditions in multi-threaded environments
• Consider implementing authentication/authorization to prevent unauthorized access to analysis progress data

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function analyze_data 77.2% similar

  Flask route handler that initiates an asynchronous data analysis process based on user query, creating a background thread to perform the analysis and returning an analysis ID for progress tracking.

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

• function get_session_status 73.0% similar

  Flask API endpoint that retrieves the current status and summary of an analysis session by its ID.

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

• function get_results 71.7% similar

  Flask route handler that retrieves and returns comprehensive analysis results for a given session, including summary data, generated files, interpretations, and execution tracking information.

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

• function get_analysis_files 71.4% similar

  Flask API endpoint that retrieves files associated with a specific analysis within a session, returning them as a JSON response.

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

• function smartstat_get_progress 70.9% similar

  Flask API endpoint that retrieves the progress status of a SmartStat analysis job by job_id, returning progress data and completed results if available.

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