Google API Client Library for Python that provides discovery-based access to hundreds of Google services with authentication, caching, and media upload/download support.
—
The Google API Python Client provides a comprehensive error handling system with structured exceptions that contain detailed information about failures, enabling robust error recovery and debugging.
Foundation exception classes that provide common error handling functionality.
class Error(Exception):
"""Base exception class for all Google API client errors."""
passExceptions related to HTTP request failures with detailed response information.
class HttpError(Error):
"""HTTP request failed with an error response from the server."""
def __init__(self, resp, content, uri=None):
"""
Initialize an HTTP error.
Args:
resp (httplib2.Response): HTTP response object
content (bytes): Response body content
uri (str, optional): URI that was requested
"""
@property
def resp(self):
"""
Get the HTTP response object.
Returns:
httplib2.Response: The HTTP response with status, headers, etc.
"""
@property
def content(self):
"""
Get the response body content.
Returns:
bytes: Raw response content
"""
@property
def error_details(self):
"""
Get structured error details from the response.
Returns:
list: List of error detail dictionaries, or empty list if none
"""
@property
def status_code(self):
"""
Get the HTTP status code.
Returns:
int: HTTP status code (e.g., 404, 500)
"""
def _get_reason(self):
"""
Extract the error reason from the response.
Returns:
str: Human-readable error reason
"""
class MediaUploadSizeError(HttpError):
"""Media upload size exceeds the allowed limits."""
pass
class BatchError(HttpError):
"""Error occurred during batch request processing."""
def __init__(self, reason, resp=None, content=None):
"""
Initialize a batch error.
Args:
reason (str): Description of the batch error
resp (httplib2.Response, optional): HTTP response object
content (bytes, optional): Response content
"""
class ResumableUploadError(HttpError):
"""Error occurred during resumable upload processing."""
passExceptions for specific API operation failures.
class InvalidJsonError(Error):
"""Response contained invalid JSON that could not be parsed."""
pass
class UnknownFileType(Error):
"""Media upload failed due to unknown or unsupported file type."""
pass
class UnknownLinkType(Error):
"""Link type unknown or unexpected."""
def __init__(self):
"""Initialize an unknown link type error."""
class UnacceptableMimeTypeError(Error):
"""That is an unacceptable mimetype for this operation."""
def __init__(self):
"""Initialize an unacceptable MIME type error."""
class InvalidChunkSizeError(Error):
"""The given chunksize is not valid."""
def __init__(self):
"""Initialize an invalid chunk size error."""
class InvalidNotificationError(Error):
"""The channel Notification is invalid."""
def __init__(self):
"""Initialize an invalid notification error."""
class UnknownApiNameOrVersion(Error):
"""No API with that name and version exists."""
def __init__(self):
"""Initialize an unknown API name or version error."""
class UnexpectedMethodError(Error):
"""Exception raised by RequestMockBuilder on unexpected calls."""
def __init__(self, methodId=None):
"""
Initialize an unexpected method error.
Args:
methodId (str, optional): Method identifier that was unexpected
"""
class UnexpectedBodyError(Error):
"""Exception raised by RequestMockBuilder on unexpected bodies."""
def __init__(self, expected, provided):
"""
Initialize an unexpected body error.
Args:
expected: Expected body content
provided: Provided body content
"""from googleapiclient import discovery
from googleapiclient.errors import HttpError
service = discovery.build('gmail', 'v1', credentials=credentials)
try:
# Attempt to get a message that might not exist
message = service.users().messages().get(
userId='me',
id='nonexistent_message_id'
).execute()
except HttpError as error:
print(f'HTTP Error {error.status_code}: {error._get_reason()}')
print(f'Response content: {error.content}')from googleapiclient.errors import HttpError
import json
try:
response = service.users().messages().list(userId='invalid_user').execute()
except HttpError as error:
print(f'Status Code: {error.status_code}')
print(f'Reason: {error._get_reason()}')
# Parse error details from response content
try:
error_content = json.loads(error.content.decode('utf-8'))
if 'error' in error_content:
error_info = error_content['error']
print(f'Error Code: {error_info.get("code")}')
print(f'Error Message: {error_info.get("message")}')
# Check for detailed error information
if 'errors' in error_info:
for detail in error_info['errors']:
print(f' - {detail.get("reason")}: {detail.get("message")}')
except (json.JSONDecodeError, UnicodeDecodeError):
print(f'Raw error content: {error.content}')from googleapiclient.errors import (
HttpError, BatchError, MediaUploadSizeError,
UnknownApiNameOrVersion, InvalidJsonError
)
try:
# Various operations that might fail
service = discovery.build('unknown-api', 'v1', credentials=credentials)
except UnknownApiNameOrVersion as error:
print(f'API not available: {error}')
# Fallback to a different API or version
except HttpError as error:
if error.status_code == 401:
print('Authentication failed - check credentials')
elif error.status_code == 403:
print('Access forbidden - check permissions/quotas')
elif error.status_code == 404:
print('Resource not found')
elif error.status_code == 429:
print('Rate limit exceeded - implement backoff')
elif error.status_code >= 500:
print('Server error - retry may help')
else:
print(f'HTTP error {error.status_code}: {error._get_reason()}')
except BatchError as error:
print(f'Batch processing failed: {error}')
except MediaUploadSizeError as error:
print(f'File too large for upload: {error}')
except InvalidJsonError as error:
print(f'Invalid JSON in response: {error}')from googleapiclient.errors import HttpError
import time
import random
def execute_with_retry(request, max_retries=3, backoff_factor=1.5):
"""
Execute a request with exponential backoff retry logic.
Args:
request: The HttpRequest to execute
max_retries (int): Maximum number of retry attempts
backoff_factor (float): Multiplier for exponential backoff
Returns:
Response object from successful request
Raises:
HttpError: When all retry attempts are exhausted
"""
last_error = None
for attempt in range(max_retries + 1):
try:
return request.execute()
except HttpError as error:
last_error = error
# Don't retry on client errors (4xx) except for specific cases
if 400 <= error.status_code < 500:
if error.status_code not in [429, 408]: # Rate limit, timeout
raise
# Don't retry on the last attempt
if attempt == max_retries:
raise
# Calculate wait time with jitter
wait_time = (backoff_factor ** attempt) + random.uniform(0, 1)
print(f'Request failed (attempt {attempt + 1}), retrying in {wait_time:.1f}s...')
time.sleep(wait_time)
# This should never be reached, but just in case
raise last_error
# Usage
try:
request = service.users().messages().list(userId='me')
response = execute_with_retry(request, max_retries=3)
print(f'Success: {len(response.get("messages", []))} messages')
except HttpError as error:
print(f'Request failed after retries: {error._get_reason()}')from googleapiclient import http
from googleapiclient.errors import HttpError, BatchError
def batch_callback(request_id, response, exception):
"""Handle individual batch request results with error processing."""
if exception is not None:
if isinstance(exception, HttpError):
print(f'Request {request_id} failed with HTTP {exception.status_code}')
print(f' Reason: {exception._get_reason()}')
# Handle specific error codes
if exception.status_code == 404:
print(f' Resource not found for request {request_id}')
elif exception.status_code == 403:
print(f' Access denied for request {request_id}')
else:
print(f'Request {request_id} failed: {exception}')
else:
print(f'Request {request_id} succeeded')
try:
batch = http.BatchHttpRequest(callback=batch_callback)
# Add requests that might fail
batch.add(service.users().messages().get(userId='me', id='valid_id'),
request_id='valid_message')
batch.add(service.users().messages().get(userId='me', id='invalid_id'),
request_id='invalid_message')
batch.execute()
except BatchError as error:
print(f'Batch execution failed: {error}')
except HttpError as error:
print(f'Batch request failed: {error._get_reason()}')from googleapiclient.errors import HttpError
import logging
class ApiErrorHandler:
"""Custom error handler for Google API operations."""
def __init__(self, logger=None):
self.logger = logger or logging.getLogger(__name__)
def handle_error(self, error, operation_name, **context):
"""
Handle API errors with logging and context.
Args:
error (Exception): The error that occurred
operation_name (str): Name of the operation that failed
**context: Additional context information
"""
if isinstance(error, HttpError):
self.logger.error(
f'{operation_name} failed: HTTP {error.status_code} - {error._get_reason()}',
extra={
'status_code': error.status_code,
'operation': operation_name,
'context': context
}
)
# Return suggested action
if error.status_code == 401:
return 'refresh_credentials'
elif error.status_code == 403:
return 'check_permissions'
elif error.status_code == 429:
return 'implement_backoff'
elif error.status_code >= 500:
return 'retry_request'
else:
return 'check_request_parameters'
else:
self.logger.error(f'{operation_name} failed: {error}', extra={
'operation': operation_name,
'context': context
})
return 'unknown_error'
# Usage
error_handler = ApiErrorHandler()
try:
messages = service.users().messages().list(userId='me').execute()
except Exception as error:
action = error_handler.handle_error(
error,
'list_messages',
user_id='me',
service='gmail'
)
if action == 'refresh_credentials':
# Implement credential refresh logic
pass
elif action == 'implement_backoff':
# Implement rate limiting backoff
passfrom googleapiclient.errors import HttpError
import json
def extract_error_details(error):
"""
Extract detailed error information from HttpError.
Args:
error (HttpError): The HTTP error to analyze
Returns:
dict: Structured error information
"""
details = {
'status_code': error.status_code,
'reason': error._get_reason(),
'errors': []
}
try:
# Parse JSON error response
content = json.loads(error.content.decode('utf-8'))
if 'error' in content:
error_info = content['error']
details['code'] = error_info.get('code')
details['message'] = error_info.get('message')
# Extract individual error details
if 'errors' in error_info:
for err in error_info['errors']:
details['errors'].append({
'domain': err.get('domain'),
'reason': err.get('reason'),
'message': err.get('message'),
'location': err.get('location'),
'location_type': err.get('locationType')
})
except (json.JSONDecodeError, UnicodeDecodeError, KeyError):
# Fallback to raw content
details['raw_content'] = error.content.decode('utf-8', errors='ignore')
return details
# Usage
try:
response = service.users().messages().get(userId='me', id='bad_id').execute()
except HttpError as error:
error_details = extract_error_details(error)
print(f'Error: {error_details["message"]}')
for err in error_details['errors']:
print(f' - {err["reason"]}: {err["message"]}')
if err['location']:
print(f' Location: {err["location"]} ({err["location_type"]})')Install with Tessl CLI
npx tessl i tessl/pypi-google-api-python-client