Get Job Status

import time
import requests

def wait_for_job_completion(job_id, api_key, poll_interval=5, max_wait=300):
    """Wait for job to complete with polling"""
    
    start_time = time.time()
    headers = {"X-API-Key": api_key}
    
    while time.time() - start_time < max_wait:
        response = requests.get(f"https://visionapi.unsiloed.ai/jobs/{job_id}", headers=headers)
        
        if response.status_code == 200:
            job = response.json()
            status = job['status']
            
            print(f"Job status: {status}")
            
            if status == 'COMPLETED':
                return job, True
            elif status == 'FAILED':
                return job, False
                
        time.sleep(poll_interval)
    
    raise TimeoutError(f"Job {job_id} did not complete within {max_wait} seconds")

# Usage
job_id = "your-job-id"
job, success = wait_for_job_completion(job_id, "your-api-key")

if success:
    print("Job completed successfully!")
else:
    print(f"Job failed: {job.get('error', 'Unknown error')}")

import time

def wait_for_job_completion(job_id, max_wait_time=300, poll_interval=5):
    """
    Poll job status until completion or timeout
    """
    start_time = time.time()
    
    while time.time() - start_time < max_wait_time:
        response = requests.get(
            f"https://api.example.com/jobs/{job_id}",
            headers=headers
        )
        
        if response.status_code == 200:
            job = response.json()
            status = job['status']
            
            print(f"Job {job_id}: {status}")
            
            if status in ['completed', 'failed', 'cancelled']:
                return job
                
            # Show progress if available
            if 'progress' in job:
                print(f"  Progress: {job['progress']['percentage']}%")
        
        time.sleep(poll_interval)
    
    raise TimeoutError(f"Job {job_id} did not complete within {max_wait_time} seconds")

# Usage
try:
    completed_job = wait_for_job_completion("job_abc123def456")
    if completed_job['status'] == 'completed':
        print("Job completed successfully!")
    else:
        print(f"Job failed: {completed_job.get('error_details', {}).get('message')}")
except TimeoutError as e:
    print(f"Timeout: {e}")

def poll_with_backoff(job_id, initial_interval=2, max_interval=30, backoff_factor=1.5):
    """
    Poll with exponential backoff to reduce API calls
    """
    interval = initial_interval
    
    while True:
        response = requests.get(f"https://api.example.com/jobs/{job_id}", headers=headers)
        job = response.json()
        
        if job['status'] in ['completed', 'failed', 'cancelled']:
            return job
            
        print(f"Job {job_id}: {job['status']} - waiting {interval}s")
        time.sleep(interval)
        
        # Increase interval for next poll
        interval = min(interval * backoff_factor, max_interval)

const ws = new WebSocket('wss://api.example.com/jobs/job_abc123def456/status');

ws.onopen = function() {
    console.log('Connected to job status stream');
};

ws.onmessage = function(event) {
    const statusUpdate = JSON.parse(event.data);
    
    console.log(`Job ${statusUpdate.job_id}: ${statusUpdate.status}`);
    
    if (statusUpdate.progress) {
        console.log(`Progress: ${statusUpdate.progress.percentage}%`);
        updateProgressBar(statusUpdate.progress.percentage);
    }
    
    if (statusUpdate.status === 'completed') {
        console.log('Job completed!');
        ws.close();
        handleJobCompletion(statusUpdate);
    } else if (statusUpdate.status === 'failed') {
        console.log('Job failed:', statusUpdate.error_details);
        ws.close();
        handleJobFailure(statusUpdate);
    }
};

ws.onerror = function(error) {
    console.error('WebSocket error:', error);
};

{
  "event": "job.status_changed",
  "job_id": "job_abc123def456",
  "previous_status": "processing",
  "current_status": "completed",
  "timestamp": "2024-01-15T10:35:42Z",
  "progress": {
    "percentage": 100,
    "files_completed": 3,
    "total_files": 3
  }
}

# Enable email notifications when creating job
data = {
    'operation': 'extract',
    'parameters': extraction_params,
    'notification_settings': {
        'email': 'user@example.com',
        'events': ['completed', 'failed'],
        'include_summary': True
    }
}

{
  "error": "Job not found",
  "details": {
    "error_type": "not_found",
    "message": "Job with ID 'job_invalid123' does not exist",
    "job_id": "job_invalid123"
  },
  "suggestions": [
    "Check the job ID for typos",
    "Verify the job belongs to your account",
    "Job may have been automatically deleted after retention period"
  ]
}

# Instead of checking each job individually
for job_id in job_ids:
    response = requests.get(f"https://api.example.com/jobs/{job_id}")  # Multiple API calls

# Use batch status check
response = requests.get(
    "https://api.example.com/jobs/status",
    params={"job_ids": ",".join(job_ids)}  # Single API call
)

import time
from functools import lru_cache

@lru_cache(maxsize=100)
def get_job_status_cached(job_id, cache_duration=30):
    """Cache job status for 30 seconds to reduce API calls"""
    cache_key = f"{job_id}_{int(time.time() // cache_duration)}"
    
    response = requests.get(f"https://api.example.com/jobs/{job_id}", headers=headers)
    return response.json()

# Usage - will use cached result if called within 30 seconds
status = get_job_status_cached("job_abc123def456")