Comprehensive exception system for handling authentication errors, application errors, and communication failures with detailed error information.
Handle authentication and authorization failures when connecting to protected Gradio applications.
class AuthenticationError(ValueError):
"""
Raised when the client is unable to authenticate itself to a Gradio app
due to invalid or missing credentials.
This exception occurs when:
- Invalid or expired Hugging Face token for private Spaces
- Missing authentication credentials for protected apps
- Token lacks required permissions for Space operations
Inherits from ValueError for standard error handling patterns.
"""
passHandle errors raised by the upstream Gradio application during prediction or processing.
class AppError(ValueError):
"""
Raised when the upstream Gradio app throws an error because of the
value submitted by the client.
This exception provides detailed error information and display options
for debugging and user interaction.
"""
def __init__(
self,
message: str = "Error raised.",
duration: float | None = 10,
visible: bool = True,
title: str = "Error",
print_exception: bool = True,
):
"""
Initialize an AppError with detailed error information.
Parameters:
- message: The error message to be displayed to the user. Can be HTML.
- duration: Duration in seconds to display error. None/0 for persistent display.
- visible: Whether the error message should be displayed in the UI.
- title: The title to be displayed at the top of the error modal.
- print_exception: Whether to print traceback to console when raised.
"""
self.title = title
self.message = message
self.duration = duration
self.visible = visible
self.print_exception = print_exception
super().__init__(self.message)Handle network and communication failures during client-server interactions.
class TooManyRequestsError(Exception):
"""
Raised when the API returns a 429 status code.
Indicates rate limiting is in effect and requests should be
throttled or retried after a delay.
"""
pass
class QueueError(Exception):
"""
Raised when the queue is full or there is an issue adding a job to the queue.
This can occur when:
- Queue is full or unavailable
- Job submission fails
- Queue processing errors
"""
pass
class InvalidAPIEndpointError(Exception):
"""
Raised when the API endpoint is invalid.
This occurs when:
- Specified api_name doesn't exist in the Gradio app
- fn_index is out of range
- Endpoint configuration is malformed
"""
pass
class SpaceDuplicationError(Exception):
"""
Raised when something goes wrong with a Space Duplication.
This can occur when:
- Source Space doesn't exist or is inaccessible
- Insufficient permissions for duplication
- Hardware requirements cannot be met
- Network issues during duplication process
"""
passHandle errors related to data serialization and deserialization.
class SerializationSetupError(ValueError):
"""
Raised when a serializers cannot be set up correctly.
This occurs when:
- Unsupported data types in predictions
- Serialization format incompatibilities
- Missing required serialization components
"""
passfrom gradio_client import Client
from gradio_client.exceptions import AuthenticationError, AppError
try:
# Attempt to connect to a private Space
client = Client("username/private-space", hf_token="invalid_token")
result = client.predict("test input")
except AuthenticationError as e:
print(f"Authentication failed: {e}")
print("Please check your Hugging Face token")
except AppError as e:
print(f"Application error: {e.message}")
print(f"Error title: {e.title}")
if e.print_exception:
import traceback
traceback.print_exc()from gradio_client import Client
from gradio_client.exceptions import (
AuthenticationError,
AppError,
TooManyRequestsError,
SerializationSetupError
)
import time
import httpx
def robust_prediction(space_id: str, input_data, max_retries: int = 3):
"""Make a prediction with comprehensive error handling and retries."""
for attempt in range(max_retries):
try:
client = Client(space_id)
return client.predict(input_data, api_name="/predict")
except AuthenticationError as e:
print(f"Authentication error: {e}")
# Don't retry authentication errors - fix credentials instead
raise
except TooManyRequestsError as e:
print(f"Rate limited (attempt {attempt + 1}/{max_retries})")
if attempt < max_retries - 1:
# Exponential backoff
wait_time = 2 ** attempt
print(f"Waiting {wait_time} seconds before retry...")
time.sleep(wait_time)
else:
raise
except AppError as e:
print(f"Application error: {e.message}")
# Log error details
print(f"Title: {e.title}")
print(f"Duration: {e.duration}")
# Don't retry application errors - fix input instead
raise
except SerializationSetupError as e:
print(f"Serialization error: {e}")
# Don't retry serialization errors - fix data format
raise
except httpx.TimeoutException as e:
print(f"Timeout error (attempt {attempt + 1}/{max_retries}): {e}")
if attempt < max_retries - 1:
time.sleep(1)
else:
raise
except Exception as e:
print(f"Unexpected error (attempt {attempt + 1}/{max_retries}): {e}")
if attempt < max_retries - 1:
time.sleep(1)
else:
raise
# Usage
try:
result = robust_prediction("abidlabs/whisper", "audio.wav")
print(f"Success: {result}")
except Exception as e:
print(f"Final error: {e}")from gradio_client import Client
from gradio_client.exceptions import AppError
import concurrent.futures
client = Client("abidlabs/batch-processor")
def process_with_timeout(input_data, timeout=30):
"""Process input with timeout and error handling."""
try:
job = client.submit(input_data, api_name="/process")
# Wait for result with timeout
result = job.result(timeout=timeout)
return {"success": True, "result": result, "error": None}
except TimeoutError:
# Try to cancel the job
cancelled = job.cancel()
return {
"success": False,
"result": None,
"error": f"Timeout after {timeout}s, cancelled: {cancelled}"
}
except AppError as e:
return {
"success": False,
"result": None,
"error": f"App error: {e.message}"
}
# Process multiple inputs
inputs = ["data1.csv", "data2.csv", "data3.csv"]
results = []
for input_data in inputs:
result = process_with_timeout(input_data, timeout=60)
results.append(result)
if not result["success"]:
print(f"Failed to process {input_data}: {result['error']}")
print(f"Successfully processed {sum(1 for r in results if r['success'])} out of {len(inputs)} items")from gradio_client import Client
from gradio_client.exceptions import AppError
import logging
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class GradioClientWrapper:
"""Wrapper with enhanced error handling and logging."""
def __init__(self, space_id: str, **kwargs):
self.space_id = space_id
try:
self.client = Client(space_id, **kwargs)
logger.info(f"Successfully connected to {space_id}")
except Exception as e:
logger.error(f"Failed to connect to {space_id}: {e}")
raise
def safe_predict(self, *args, **kwargs):
"""Make prediction with detailed error context."""
try:
logger.info(f"Making prediction with args: {args}")
result = self.client.predict(*args, **kwargs)
logger.info("Prediction successful")
return result
except AppError as e:
# Log application-specific errors with context
logger.error(f"Application error in {self.space_id}")
logger.error(f"Title: {e.title}")
logger.error(f"Message: {e.message}")
logger.error(f"Duration: {e.duration}")
# Create custom exception with more context
raise RuntimeError(
f"Prediction failed for Space '{self.space_id}': {e.message}"
) from e
except Exception as e:
logger.error(f"Unexpected error in {self.space_id}: {e}")
raise RuntimeError(
f"Unexpected error in Space '{self.space_id}': {str(e)}"
) from e
# Usage
try:
wrapper = GradioClientWrapper("abidlabs/whisper")
result = wrapper.safe_predict("audio.wav", api_name="/predict")
print(f"Transcription: {result}")
except RuntimeError as e:
print(f"Processing failed: {e}")
# Original exception is available as e.__cause__from gradio_client import Client
from gradio_client.exceptions import AuthenticationError, AppError
import warnings
def get_client_with_fallback(primary_space: str, fallback_space: str, hf_token: str | None = None):
"""Get client with fallback to alternative space."""
# Try primary space first
try:
client = Client(primary_space, hf_token=hf_token)
print(f"Connected to primary space: {primary_space}")
return client, primary_space
except AuthenticationError:
warnings.warn(f"Authentication failed for {primary_space}, trying fallback")
except Exception as e:
warnings.warn(f"Failed to connect to {primary_space}: {e}, trying fallback")
# Try fallback space
try:
client = Client(fallback_space, hf_token=hf_token)
print(f"Connected to fallback space: {fallback_space}")
return client, fallback_space
except Exception as e:
raise RuntimeError(f"Failed to connect to both {primary_space} and {fallback_space}") from e
def predict_with_fallback(primary_space: str, fallback_space: str, input_data, **kwargs):
"""Make prediction with automatic fallback."""
client, active_space = get_client_with_fallback(primary_space, fallback_space)
try:
result = client.predict(input_data, **kwargs)
return result, active_space
except AppError as e:
if active_space == primary_space:
warnings.warn(f"Primary space failed: {e.message}, trying fallback")
# Try fallback
try:
fallback_client = Client(fallback_space)
result = fallback_client.predict(input_data, **kwargs)
return result, fallback_space
except Exception as fallback_error:
raise RuntimeError(f"Both spaces failed. Primary: {e.message}, Fallback: {fallback_error}") from e
else:
# Already using fallback, reraise
raise
# Usage
try:
result, used_space = predict_with_fallback(
"user/private-whisper",
"abidlabs/whisper",
"audio.wav",
api_name="/predict"
)
print(f"Transcription from {used_space}: {result}")
except RuntimeError as e:
print(f"All options exhausted: {e}")