🔍 Code Extractor

Flask route handler that retrieves and returns the current progress status of an enhanced SQL workflow, including step completion, progress percentage, and final results if completed.

Purpose

This endpoint provides real-time progress tracking for enhanced SQL workflows. It checks the application's workflow storage, determines the current step and progress percentage, maps step statuses (waiting/active/completed), and returns comprehensive progress information including iteration counts, quality scores, and final results when the workflow is complete. Used for polling-based progress updates in the frontend.

Source Code

def enhanced_workflow_progress(workflow_id):
    """Get progress of enhanced workflow"""
    try:
        # Check if we have stored workflow status
        workflows = getattr(app, 'enhanced_workflows', {})
        workflow_data = workflows.get(workflow_id)
        
        if not workflow_data:
            return jsonify({
                'success': False,
                'error': 'Workflow not found'
            }), 404
        
        current_step = workflow_data.get('current_step', 'parsing')
        status = workflow_data.get('status', 'in_progress')
        progress = workflow_data.get('progress', 0)
        
        # Map current step to step status
        steps = {
            'parsing': 'waiting',
            'query_generation': 'waiting',
            'data_retrieval': 'waiting', 
            'quality_assessment': 'waiting',
            'optimization': 'waiting',
            'finalization': 'waiting'
        }
        
        # Update step statuses based on current progress
        if current_step == 'parsing' or progress >= 10:
            steps['parsing'] = 'completed' if progress > 10 else 'active'
        if current_step == 'query_generation' or progress >= 30:
            steps['query_generation'] = 'completed' if progress > 30 else 'active'
        if current_step == 'data_retrieval' or progress >= 50:
            steps['data_retrieval'] = 'completed' if progress > 50 else 'active'
        if current_step == 'quality_assessment' or progress >= 70:
            steps['quality_assessment'] = 'completed' if progress > 70 else 'active'
        if current_step == 'optimization' or progress >= 80:
            steps['optimization'] = 'completed' if progress > 80 else 'active'
        if current_step == 'finalization' or progress >= 90:
            steps['finalization'] = 'completed' if progress >= 100 else 'active'
        
        response = {
            'success': True,
            'workflow_id': workflow_id,
            'status': status,
            'current_step': current_step,
            'progress': progress,
            'steps': steps,
            'message': workflow_data.get('message', 'Processing...'),
            'current_iteration': len(workflow_data.get('iterations', [])) or 1
        }
        
        # Add completion details if finished
        if status == 'completed' and 'result' in workflow_data:
            result = workflow_data['result']
            response['final_results'] = {
                'final_rows': result['dataframe'].shape[0] if 'dataframe' in result and result['dataframe'] is not None else 0,
                'final_columns': result['dataframe'].shape[1] if 'dataframe' in result and result['dataframe'] is not None else 0,
                'quality_score': result.get('metadata', {}).get('final_quality_score', 100),
                'iterations_used': len(result.get('iterations', [])) or 1,
                'optimization_achieved': result.get('metadata', {}).get('optimization_achieved', True),
                'analysis_ready': True
            }
        
        # Add error details if failed
        if status == 'failed':
            response['error'] = workflow_data.get('error', 'Unknown error')
        
        return jsonify(response)
        
    except Exception as e:
        logger.error(f"Error getting workflow progress: {str(e)}")
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

Parameters

Parameter Details

workflow_id: String identifier for the workflow to track. This ID is used to look up the workflow data from the app.enhanced_workflows dictionary. Must match a previously initiated workflow ID or will return a 404 error.

Return Value

Returns a Flask JSON response tuple. On success (200): JSON object with 'success': True, 'workflow_id', 'status' (in_progress/completed/failed), 'current_step' (parsing/query_generation/data_retrieval/quality_assessment/optimization/finalization), 'progress' (0-100), 'steps' (dict mapping step names to status), 'message', 'current_iteration', and optionally 'final_results' (with final_rows, final_columns, quality_score, iterations_used, optimization_achieved, analysis_ready) or 'error' details. On failure: JSON with 'success': False and 'error' message, with status code 404 (not found) or 500 (server error).

Dependencies

• flask
• logging

Required Imports

from flask import jsonify
import logging

Usage Example

# Assuming Flask app is set up with workflow storage
# Client-side polling example:
import requests
import time

workflow_id = 'abc-123-def-456'
base_url = 'http://localhost:5000'

while True:
    response = requests.get(f'{base_url}/enhanced_workflow_progress/{workflow_id}')
    data = response.json()
    
    if data['success']:
        print(f"Progress: {data['progress']}%")
        print(f"Current step: {data['current_step']}")
        print(f"Status: {data['status']}")
        
        if data['status'] == 'completed':
            print(f"Final results: {data['final_results']}")
            break
        elif data['status'] == 'failed':
            print(f"Error: {data['error']}")
            break
    else:
        print(f"Error: {data['error']}")
        break
    
    time.sleep(2)  # Poll every 2 seconds

Best Practices

• This endpoint is designed for polling; implement appropriate polling intervals (2-5 seconds) to avoid overwhelming the server
• The workflow_id must be stored in app.enhanced_workflows before calling this endpoint
• Workflow data structure should include: 'current_step', 'status', 'progress', 'message', 'iterations', and optionally 'result' or 'error'
• Progress values should be between 0-100, with specific thresholds (10, 30, 50, 70, 80, 90, 100) corresponding to step transitions
• The 'result' object in workflow_data should contain a 'dataframe' key with a pandas DataFrame and a 'metadata' dict with quality scores
• Handle both completed and failed states appropriately on the client side
• Consider implementing WebSocket or Server-Sent Events for more efficient real-time updates instead of polling
• Ensure proper error logging is configured before using this endpoint
• Clean up completed/failed workflows from app.enhanced_workflows periodically to prevent memory leaks

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function enhanced_sql_workflow 78.1% similar

  Flask route handler that initiates an enhanced SQL workflow with iterative optimization, executing data extraction and analysis in a background thread while providing real-time progress tracking.

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

• function get_analysis_progress 69.3% similar

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

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

• function api_index_progress 67.4% similar

  Flask API endpoint that retrieves the current progress status of an asynchronous indexing task by its task ID.

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

• function text_chat_get_progress 67.2% similar

  Flask API endpoint that retrieves the progress status of an asynchronous text chat job, particularly for multi-cycle web search operations.

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

• function smartstat_get_progress 67.2% 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