abdullahceylan-ac-comfyui-queue-manager/docs/EXAMPLES.md

ComfyUI Queue Manager - Examples and Use Cases

Table of Contents

1. Basic Usage Examples
2. Workflow Management
3. Automation Scripts
4. Integration Examples
5. Advanced Use Cases
6. Troubleshooting Examples

Basic Usage Examples

Example 1: Simple Image Processing Workflow

This example shows how to create and manage a basic image processing workflow.

Workflow Setup:

1. Create a workflow in ComfyUI:

   • Load Image node
   • Image processing nodes (resize, enhance, etc.)
   • Save Image node

2. Execute the workflow normally

3. View in Queue Manager

Expected Queue Manager Behavior:

• Workflow appears as "pending"
• Status changes to "running" during execution
• Status changes to "completed" when finished
• Results are stored and viewable

Example 2: Batch Processing Multiple Images

Scenario: Process 50 images with the same workflow

Steps:

1. Create base workflow for single image
2. Use ComfyUI's batch processing features
3. Queue multiple executions
4. Monitor progress in Queue Manager

Queue Manager Features Used:

• Filter by status to see progress
• Bulk operations for completed items
• Archive processed batches

Example 3: AI Art Generation Pipeline

Scenario: Generate multiple variations of an art piece

Workflow Components:

• Text prompt input
• Model loading
• Sampling with different seeds
• Upscaling
• Final output

Queue Management:

# Example API usage for batch generation
import requests
import json

BASE_URL = "http://localhost:5000/api/queue"

# Base workflow template
base_workflow = {
    "nodes": [
        {"id": "1", "type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "model.safetensors"}},
        {"id": "2", "type": "CLIPTextEncode", "inputs": {"text": "a beautiful landscape", "clip": ["1", 1]}},
        {"id": "3", "type": "KSampler", "inputs": {"seed": 42, "steps": 20, "model": ["1", 0], "positive": ["2", 0]}},
        {"id": "4", "type": "VAEDecode", "inputs": {"samples": ["3", 0], "vae": ["1", 2]}},
        {"id": "5", "type": "SaveImage", "inputs": {"images": ["4", 0]}}
    ],
    "links": [["1", 0, "3", 0], ["1", 1, "2", 0], ["1", 2, "4", 1], ["2", 0, "3", 1], ["3", 0, "4", 0], ["4", 0, "5", 0]]
}

# Generate 10 variations with different seeds
for i in range(10):
    workflow = base_workflow.copy()
    workflow["nodes"][2]["inputs"]["seed"] = 42 + i  # Different seed for each
    
    response = requests.post(
        f"{BASE_URL}/items",
        headers={"Content-Type": "application/json"},
        data=json.dumps({
            "workflow_name": f"Art Generation Variation {i+1}",
            "workflow_data": workflow
        })
    )
    
    print(f"Queued variation {i+1}")

# Monitor progress
response = requests.get(f"{BASE_URL}/status")
status = response.json()
print(f"Queue status: {status['pending_items']} pending, {status['running_items']} running")

Workflow Management

Managing Long-Running Workflows

Scenario: Video processing or complex AI workflows that take hours

Best Practices:

1. Monitor Progress: Use Queue Manager to track status
2. Resource Management: Limit concurrent workflows
3. Error Handling: Set up retry mechanisms
4. Archiving: Archive completed long workflows

Configuration Example:

# Set maximum concurrent workflows to 1 for resource-intensive tasks
config_update = {
    "max_concurrent_workflows": 1,
    "auto_archive_completed": True,
    "auto_archive_days": 7
}

response = requests.put(
    f"{BASE_URL}/config",
    headers={"Content-Type": "application/json"},
    data=json.dumps(config_update)
)

Workflow Templates and Reuse

Scenario: Standardize workflows for team use

Implementation:

1. Create template workflows
2. Export successful workflows
3. Import templates on other systems
4. Version control workflow templates

Export Template Example:

# Export specific successful workflows as templates
template_items = ["uuid-1", "uuid-2", "uuid-3"]  # IDs of template workflows

export_data = {
    "item_ids": template_items,
    "include_archived": False
}

response = requests.post(
    f"{BASE_URL}/export",
    headers={"Content-Type": "application/json"},
    data=json.dumps(export_data)
)

# Save as template file
with open("workflow_templates.json", "w") as f:
    json.dump(response.json()["export_data"], f, indent=2)

Quality Control Workflow

Scenario: Review and approve generated content

Process:

1. Generate content with workflows
2. Filter completed items for review
3. Use custom status or tags for approval
4. Archive approved items separately

Implementation:

# Get completed items for review
filter_data = {"status": ["completed"]}
response = requests.post(f"{BASE_URL}/items/filter", 
                        headers={"Content-Type": "application/json"},
                        data=json.dumps(filter_data))

completed_items = response.json()["items"]

# Review process (manual or automated)
approved_items = []
for item in completed_items:
    # Review logic here
    if review_item(item):  # Your review function
        approved_items.append(item["id"])

# Archive approved items
if approved_items:
    archive_data = {"item_ids": approved_items}
    requests.post(f"{BASE_URL}/archive",
                 headers={"Content-Type": "application/json"},
                 data=json.dumps(archive_data))

Automation Scripts

Daily Batch Processing Script

Use Case: Process daily uploads automatically

#!/usr/bin/env python3
"""
Daily batch processing script for ComfyUI Queue Manager
"""

import requests
import json
import os
import glob
from datetime import datetime

BASE_URL = "http://localhost:5000/api/queue"
INPUT_DIR = "/path/to/daily/inputs"
WORKFLOW_TEMPLATE = "daily_processing_template.json"

def load_workflow_template():
    """Load workflow template from file"""
    with open(WORKFLOW_TEMPLATE, 'r') as f:
        return json.load(f)

def process_daily_batch():
    """Process all files in input directory"""
    # Get list of files to process
    image_files = glob.glob(os.path.join(INPUT_DIR, "*.jpg")) + \
                  glob.glob(os.path.join(INPUT_DIR, "*.png"))
    
    if not image_files:
        print("No files to process")
        return
    
    # Load workflow template
    template = load_workflow_template()
    
    # Queue workflows for each file
    queued_count = 0
    for image_file in image_files:
        workflow = template.copy()
        
        # Update workflow with specific file
        workflow["nodes"][0]["inputs"]["image"] = os.path.basename(image_file)
        
        # Queue the workflow
        response = requests.post(
            f"{BASE_URL}/items",
            headers={"Content-Type": "application/json"},
            data=json.dumps({
                "workflow_name": f"Daily Processing - {os.path.basename(image_file)}",
                "workflow_data": workflow
            })
        )
        
        if response.status_code == 201:
            queued_count += 1
            print(f"Queued: {os.path.basename(image_file)}")
        else:
            print(f"Failed to queue: {os.path.basename(image_file)}")
    
    print(f"Queued {queued_count} workflows for processing")

def cleanup_old_workflows():
    """Archive workflows older than 7 days"""
    from datetime import datetime, timedelta
    
    # Get all completed workflows
    response = requests.post(
        f"{BASE_URL}/items/filter",
        headers={"Content-Type": "application/json"},
        data=json.dumps({"status": ["completed"]})
    )
    
    completed_items = response.json()["items"]
    
    # Find items older than 7 days
    cutoff_date = datetime.now() - timedelta(days=7)
    old_items = []
    
    for item in completed_items:
        item_date = datetime.fromisoformat(item["completed_at"].replace('Z', '+00:00'))
        if item_date < cutoff_date:
            old_items.append(item["id"])
    
    # Archive old items
    if old_items:
        archive_data = {"item_ids": old_items}
        response = requests.post(
            f"{BASE_URL}/archive",
            headers={"Content-Type": "application/json"},
            data=json.dumps(archive_data)
        )
        print(f"Archived {len(old_items)} old workflows")

if __name__ == "__main__":
    print(f"Starting daily batch processing - {datetime.now()}")
    process_daily_batch()
    cleanup_old_workflows()
    print("Daily batch processing completed")

Monitoring and Alerting Script

Use Case: Monitor queue health and send alerts

#!/usr/bin/env python3
"""
Queue monitoring and alerting script
"""

import requests
import smtplib
from email.mime.text import MimeText
from datetime import datetime, timedelta

BASE_URL = "http://localhost:5000/api/queue"

def check_queue_health():
    """Check queue health and return status"""
    try:
        response = requests.get(f"{BASE_URL}/status", timeout=10)
        if response.status_code != 200:
            return {"healthy": False, "error": f"API returned {response.status_code}"}
        
        status = response.json()
        
        # Check for issues
        issues = []
        
        # Too many failed items
        if status["failed_items"] > 10:
            issues.append(f"High number of failed items: {status['failed_items']}")
        
        # Queue stuck (no progress in last hour)
        if status["running_items"] == 0 and status["pending_items"] > 0:
            issues.append("Queue appears stuck - pending items but none running")
        
        # Very old pending items
        response = requests.post(
            f"{BASE_URL}/items/filter",
            headers={"Content-Type": "application/json"},
            data=json.dumps({"status": ["pending"]})
        )
        
        if response.status_code == 200:
            pending_items = response.json()["items"]
            old_pending = []
            cutoff = datetime.now() - timedelta(hours=2)
            
            for item in pending_items:
                created_at = datetime.fromisoformat(item["created_at"].replace('Z', '+00:00'))
                if created_at < cutoff:
                    old_pending.append(item)
            
            if len(old_pending) > 5:
                issues.append(f"Many old pending items: {len(old_pending)}")
        
        return {
            "healthy": len(issues) == 0,
            "status": status,
            "issues": issues
        }
        
    except Exception as e:
        return {"healthy": False, "error": str(e)}

def send_alert(subject, message):
    """Send email alert"""
    # Configure your email settings
    smtp_server = "smtp.gmail.com"
    smtp_port = 587
    sender_email = "your-email@gmail.com"
    sender_password = "your-app-password"
    recipient_email = "admin@yourcompany.com"
    
    msg = MimeText(message)
    msg['Subject'] = subject
    msg['From'] = sender_email
    msg['To'] = recipient_email
    
    try:
        server = smtplib.SMTP(smtp_server, smtp_port)
        server.starttls()
        server.login(sender_email, sender_password)
        server.send_message(msg)
        server.quit()
        print("Alert sent successfully")
    except Exception as e:
        print(f"Failed to send alert: {e}")

def main():
    """Main monitoring function"""
    health = check_queue_health()
    
    if not health["healthy"]:
        subject = "ComfyUI Queue Manager Alert"
        
        if "error" in health:
            message = f"Queue Manager health check failed: {health['error']}"
        else:
            message = f"Queue Manager issues detected:\n\n"
            for issue in health["issues"]:
                message += f"- {issue}\n"
            
            if "status" in health:
                status = health["status"]
                message += f"\nCurrent Status:\n"
                message += f"- Total items: {status['total_items']}\n"
                message += f"- Pending: {status['pending_items']}\n"
                message += f"- Running: {status['running_items']}\n"
                message += f"- Failed: {status['failed_items']}\n"
        
        send_alert(subject, message)
    else:
        print("Queue health check passed")

if __name__ == "__main__":
    main()

Integration Examples

Integration with External Systems

Webhook Integration

Use Case: Notify external systems when workflows complete

import requests
import json

def setup_webhook_monitoring():
    """Monitor queue and send webhooks for completed workflows"""
    
    # Get recently completed items
    response = requests.post(
        f"{BASE_URL}/items/filter",
        headers={"Content-Type": "application/json"},
        data=json.dumps({
            "status": ["completed"],
            "date_from": (datetime.now() - timedelta(hours=1)).isoformat()
        })
    )
    
    completed_items = response.json()["items"]
    
    # Send webhook for each completed item
    webhook_url = "https://your-system.com/webhook/comfyui-completed"
    
    for item in completed_items:
        webhook_data = {
            "event": "workflow_completed",
            "workflow_id": item["id"],
            "workflow_name": item["workflow_name"],
            "completed_at": item["completed_at"],
            "result_data": item["result_data"]
        }
        
        try:
            response = requests.post(
                webhook_url,
                headers={"Content-Type": "application/json"},
                data=json.dumps(webhook_data),
                timeout=10
            )
            print(f"Webhook sent for {item['id']}: {response.status_code}")
        except Exception as e:
            print(f"Webhook failed for {item['id']}: {e}")

Database Integration

Use Case: Store workflow results in external database

import sqlite3
import requests
import json

def sync_to_database():
    """Sync completed workflows to external database"""
    
    # Connect to external database
    conn = sqlite3.connect('workflow_results.db')
    cursor = conn.cursor()
    
    # Create table if not exists
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS workflow_results (
            id TEXT PRIMARY KEY,
            workflow_name TEXT,
            status TEXT,
            created_at TEXT,
            completed_at TEXT,
            result_data TEXT,
            synced_at TEXT
        )
    ''')
    
    # Get completed workflows not yet synced
    response = requests.post(
        f"{BASE_URL}/items/filter",
        headers={"Content-Type": "application/json"},
        data=json.dumps({"status": ["completed"]})
    )
    
    completed_items = response.json()["items"]
    
    # Check which items are already synced
    synced_count = 0
    for item in completed_items:
        cursor.execute('SELECT id FROM workflow_results WHERE id = ?', (item["id"],))
        if cursor.fetchone() is None:
            # Insert new record
            cursor.execute('''
                INSERT INTO workflow_results 
                (id, workflow_name, status, created_at, completed_at, result_data, synced_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            ''', (
                item["id"],
                item["workflow_name"],
                item["status"],
                item["created_at"],
                item["completed_at"],
                json.dumps(item["result_data"]),
                datetime.now().isoformat()
            ))
            synced_count += 1
    
    conn.commit()
    conn.close()
    print(f"Synced {synced_count} new workflow results to database")

CI/CD Integration

Use Case: Automated testing of ComfyUI workflows

# .github/workflows/test-workflows.yml
name: Test ComfyUI Workflows

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test-workflows:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v2
    
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.9'
    
    - name: Install dependencies
      run: |
        pip install requests
    
    - name: Start ComfyUI with Queue Manager
      run: |
        # Start ComfyUI in background
        python ComfyUI/main.py --headless &
        sleep 30  # Wait for startup
    
    - name: Test Workflow Execution
      run: |
        python test_workflows.py
    
    - name: Check Results
      run: |
        python check_test_results.py

# test_workflows.py
import requests
import json
import time

BASE_URL = "http://localhost:5000/api/queue"

def test_basic_workflow():
    """Test basic image processing workflow"""
    
    workflow_data = {
        "workflow_name": "CI Test - Basic Processing",
        "workflow_data": {
            "nodes": [
                {"id": "1", "type": "LoadImage", "inputs": {"image": "test_image.png"}},
                {"id": "2", "type": "SaveImage", "inputs": {"images": ["1", 0]}}
            ],
            "links": [["1", 0, "2", 0]]
        }
    }
    
    # Queue workflow
    response = requests.post(
        f"{BASE_URL}/items",
        headers={"Content-Type": "application/json"},
        data=json.dumps(workflow_data)
    )
    
    assert response.status_code == 201
    item_id = response.json()["item"]["id"]
    
    # Wait for completion (with timeout)
    timeout = 300  # 5 minutes
    start_time = time.time()
    
    while time.time() - start_time < timeout:
        response = requests.get(f"{BASE_URL}/items/{item_id}")
        item = response.json()["item"]
        
        if item["status"] == "completed":
            print("✓ Basic workflow test passed")
            return True
        elif item["status"] == "failed":
            print(f"✗ Basic workflow test failed: {item['error_message']}")
            return False
        
        time.sleep(5)
    
    print("✗ Basic workflow test timed out")
    return False

if __name__ == "__main__":
    success = test_basic_workflow()
    exit(0 if success else 1)

Advanced Use Cases

Multi-Stage Processing Pipeline

Scenario: Complex pipeline with multiple dependent stages

class WorkflowPipeline:
    """Manage multi-stage workflow pipeline"""
    
    def __init__(self, base_url):
        self.base_url = base_url
        self.stages = []
    
    def add_stage(self, name, workflow_template, depends_on=None):
        """Add a stage to the pipeline"""
        self.stages.append({
            "name": name,
            "workflow_template": workflow_template,
            "depends_on": depends_on,
            "items": []
        })
    
    def execute_pipeline(self, input_data):
        """Execute the entire pipeline"""
        pipeline_id = f"pipeline_{int(time.time())}"
        
        for stage in self.stages:
            if stage["depends_on"]:
                # Wait for dependency to complete
                self.wait_for_stage_completion(stage["depends_on"])
            
            # Execute current stage
            workflow = stage["workflow_template"].copy()
            # Update workflow with input_data and previous stage results
            
            response = requests.post(
                f"{self.base_url}/items",
                headers={"Content-Type": "application/json"},
                data=json.dumps({
                    "workflow_name": f"{pipeline_id} - {stage['name']}",
                    "workflow_data": workflow
                })
            )
            
            if response.status_code == 201:
                item_id = response.json()["item"]["id"]
                stage["items"].append(item_id)
                print(f"Started stage: {stage['name']}")
    
    def wait_for_stage_completion(self, stage_name):
        """Wait for a specific stage to complete"""
        stage = next(s for s in self.stages if s["name"] == stage_name)
        
        while True:
            all_completed = True
            for item_id in stage["items"]:
                response = requests.get(f"{self.base_url}/items/{item_id}")
                item = response.json()["item"]
                
                if item["status"] not in ["completed", "failed"]:
                    all_completed = False
                    break
            
            if all_completed:
                break
            
            time.sleep(10)

# Usage example
pipeline = WorkflowPipeline(BASE_URL)

# Stage 1: Initial processing
pipeline.add_stage("preprocessing", {
    "nodes": [
        {"id": "1", "type": "LoadImage", "inputs": {}},
        {"id": "2", "type": "ImageResize", "inputs": {"image": ["1", 0]}}
    ]
})

# Stage 2: AI processing (depends on stage 1)
pipeline.add_stage("ai_processing", {
    "nodes": [
        {"id": "1", "type": "LoadProcessedImage", "inputs": {}},
        {"id": "2", "type": "AIEnhancement", "inputs": {"image": ["1", 0]}}
    ]
}, depends_on="preprocessing")

# Stage 3: Final output (depends on stage 2)
pipeline.add_stage("output", {
    "nodes": [
        {"id": "1", "type": "LoadEnhancedImage", "inputs": {}},
        {"id": "2", "type": "SaveFinalImage", "inputs": {"image": ["1", 0]}}
    ]
}, depends_on="ai_processing")

pipeline.execute_pipeline({"input_image": "source.jpg"})

Resource-Aware Scheduling

Use Case: Optimize workflow execution based on system resources

import psutil
import requests
import json
import time

class ResourceAwareScheduler:
    """Schedule workflows based on system resources"""
    
    def __init__(self, base_url):
        self.base_url = base_url
        self.max_cpu_percent = 80
        self.max_memory_percent = 85
        self.max_concurrent = 3
    
    def get_system_resources(self):
        """Get current system resource usage"""
        return {
            "cpu_percent": psutil.cpu_percent(interval=1),
            "memory_percent": psutil.virtual_memory().percent,
            "disk_usage": psutil.disk_usage('/').percent
        }
    
    def can_start_workflow(self):
        """Check if system can handle another workflow"""
        resources = self.get_system_resources()
        
        # Check resource limits
        if resources["cpu_percent"] > self.max_cpu_percent:
            return False, "CPU usage too high"
        
        if resources["memory_percent"] > self.max_memory_percent:
            return False, "Memory usage too high"
        
        # Check current running workflows
        response = requests.get(f"{self.base_url}/status")
        status = response.json()
        
        if status["running_items"] >= self.max_concurrent:
            return False, "Too many concurrent workflows"
        
        return True, "Resources available"
    
    def schedule_workflows(self):
        """Main scheduling loop"""
        while True:
            # Get pending workflows
            response = requests.post(
                f"{self.base_url}/items/filter",
                headers={"Content-Type": "application/json"},
                data=json.dumps({"status": ["pending"]})
            )
            
            pending_items = response.json()["items"]
            
            if not pending_items:
                print("No pending workflows")
                time.sleep(30)
                continue
            
            # Check if we can start more workflows
            can_start, reason = self.can_start_workflow()
            
            if can_start:
                # Start highest priority workflow
                # (In this example, oldest first)
                oldest_item = min(pending_items, key=lambda x: x["created_at"])
                
                # Update status to running (this would normally be done by ComfyUI)
                response = requests.put(
                    f"{self.base_url}/items/{oldest_item['id']}",
                    headers={"Content-Type": "application/json"},
                    data=json.dumps({"status": "running"})
                )
                
                print(f"Started workflow: {oldest_item['workflow_name']}")
            else:
                print(f"Cannot start workflow: {reason}")
            
            time.sleep(10)  # Check every 10 seconds

# Usage
scheduler = ResourceAwareScheduler(BASE_URL)
scheduler.schedule_workflows()

Troubleshooting Examples

Common Issues and Solutions

Issue: Workflows Stuck in Pending

Diagnosis Script:

def diagnose_stuck_workflows():
    """Diagnose why workflows are stuck in pending"""
    
    # Get pending workflows
    response = requests.post(
        f"{BASE_URL}/items/filter",
        headers={"Content-Type": "application/json"},
        data=json.dumps({"status": ["pending"]})
    )
    
    pending_items = response.json()["items"]
    
    if not pending_items:
        print("No pending workflows found")
        return
    
    print(f"Found {len(pending_items)} pending workflows")
    
    # Check for very old pending items
    old_items = []
    cutoff = datetime.now() - timedelta(hours=1)
    
    for item in pending_items:
        created_at = datetime.fromisoformat(item["created_at"].replace('Z', '+00:00'))
        if created_at < cutoff:
            old_items.append(item)
    
    if old_items:
        print(f"Found {len(old_items)} workflows pending for over 1 hour:")
        for item in old_items[:5]:  # Show first 5
            print(f"  - {item['workflow_name']} (created: {item['created_at']})")
    
    # Check queue status
    response = requests.get(f"{BASE_URL}/status")
    status = response.json()
    
    print(f"\nQueue Status:")
    print(f"  - State: {status.get('queue_state', 'unknown')}")
    print(f"  - Running: {status.get('running_items', 0)}")
    print(f"  - Pending: {status.get('pending_items', 0)}")
    
    # Recommendations
    print(f"\nRecommendations:")
    if status.get('queue_state') == 'paused':
        print("  - Queue is paused. Resume queue processing.")
    elif status.get('running_items', 0) == 0:
        print("  - No workflows running. Check ComfyUI status.")
    elif len(old_items) > 0:
        print("  - Consider restarting ComfyUI or clearing stuck workflows.")

diagnose_stuck_workflows()

Issue: High Memory Usage

Memory Cleanup Script:

def cleanup_memory_usage():
    """Clean up queue to reduce memory usage"""
    
    # Archive old completed workflows
    cutoff_date = datetime.now() - timedelta(days=7)
    
    response = requests.post(
        f"{BASE_URL}/items/filter",
        headers={"Content-Type": "application/json"},
        data=json.dumps({"status": ["completed"]})
    )
    
    completed_items = response.json()["items"]
    old_completed = []
    
    for item in completed_items:
        completed_at = datetime.fromisoformat(item["completed_at"].replace('Z', '+00:00'))
        if completed_at < cutoff_date:
            old_completed.append(item["id"])
    
    if old_completed:
        print(f"Archiving {len(old_completed)} old completed workflows...")
        archive_data = {"item_ids": old_completed}
        response = requests.post(
            f"{BASE_URL}/archive",
            headers={"Content-Type": "application/json"},
            data=json.dumps(archive_data)
        )
        print("Archive completed")
    
    # Delete old failed workflows
    response = requests.post(
        f"{BASE_URL}/items/filter",
        headers={"Content-Type": "application/json"},
        data=json.dumps({"status": ["failed"]})
    )
    
    failed_items = response.json()["items"]
    old_failed = []
    
    for item in failed_items:
        created_at = datetime.fromisoformat(item["created_at"].replace('Z', '+00:00'))
        if created_at < cutoff_date:
            old_failed.append(item["id"])
    
    if old_failed:
        print(f"Deleting {len(old_failed)} old failed workflows...")
        delete_data = {"item_ids": old_failed}
        response = requests.delete(
            f"{BASE_URL}/items/bulk",
            headers={"Content-Type": "application/json"},
            data=json.dumps(delete_data)
        )
        print("Deletion completed")

cleanup_memory_usage()

Issue: Database Corruption

Database Recovery Script:

def recover_database():
    """Attempt to recover from database issues"""
    
    print("Attempting database recovery...")
    
    # First, try to export current data
    try:
        response = requests.post(
            f"{BASE_URL}/export",
            headers={"Content-Type": "application/json"},
            data=json.dumps({"include_archived": True})
        )
        
        if response.status_code == 200:
            # Save backup
            backup_file = f"queue_backup_{int(time.time())}.json"
            with open(backup_file, 'w') as f:
                json.dump(response.json()["export_data"], f, indent=2)
            print(f"Backup saved to {backup_file}")
        else:
            print("Could not create backup - database may be corrupted")
    
    except Exception as e:
        print(f"Backup failed: {e}")
    
    # Instructions for manual recovery
    print("\nManual recovery steps:")
    print("1. Stop ComfyUI")
    print("2. Backup current database file: queue_manager.db")
    print("3. Delete corrupted database file")
    print("4. Restart ComfyUI (new database will be created)")
    print("5. Import backup using the web interface or API")

recover_database()

These examples demonstrate the flexibility and power of the ComfyUI Queue Manager for various use cases, from simple workflow management to complex automation and integration scenarios.