tessl/pypi-google-api-core

Google API client core library providing common helpers, utilities, and components for Python client libraries

—

Pending

Overview

Eval results

Files

Exception Handling

Name: tessl/pypi-google-api-core
Author: tessl

Comprehensive exception hierarchy that maps HTTP and gRPC status codes to specific Python exceptions, providing consistent error handling across all Google API client libraries.

Capabilities

Base Exception Classes

Root exception classes that serve as the foundation for all Google API client errors.

class GoogleAPIError(Exception):
    """Base class for all Google API client exceptions."""

class DuplicateCredentialArgs(GoogleAPIError):
    """Raised when multiple credentials are provided to a function or method."""

class RetryError(GoogleAPIError):
    """Raised when a function has exhausted all of its available retries."""
    
class GoogleAPICallError(GoogleAPIError):
    """Base class for exceptions raised by calling API methods."""

HTTP Status Exception Hierarchy

Exception classes that map HTTP status codes to specific Python exceptions, maintaining the standard HTTP status code structure.

# 3xx Redirection
class Redirection(GoogleAPICallError):
    """Base class for HTTP 3xx redirection responses."""

class MovedPermanently(Redirection):
    """HTTP 301 Moved Permanently."""

class NotModified(Redirection):
    """HTTP 304 Not Modified."""

class TemporaryRedirect(Redirection):
    """HTTP 307 Temporary Redirect."""

class ResumeIncomplete(Redirection):
    """HTTP 308 Resume Incomplete."""

# 4xx Client Error
class ClientError(GoogleAPICallError):
    """Base class for HTTP 4xx client error responses."""

class BadRequest(ClientError):
    """HTTP 400 Bad Request."""

class InvalidArgument(BadRequest):
    """gRPC INVALID_ARGUMENT error."""

class FailedPrecondition(BadRequest):
    """gRPC FAILED_PRECONDITION error."""

class OutOfRange(BadRequest):
    """gRPC OUT_OF_RANGE error."""

class Unauthorized(ClientError):
    """HTTP 401 Unauthorized."""

class Unauthenticated(Unauthorized):
    """gRPC UNAUTHENTICATED error."""

class Forbidden(ClientError):
    """HTTP 403 Forbidden."""

class PermissionDenied(Forbidden):
    """gRPC PERMISSION_DENIED error."""

class NotFound(ClientError):
    """HTTP 404 Not Found or gRPC NOT_FOUND error."""

class MethodNotAllowed(ClientError):
    """HTTP 405 Method Not Allowed."""

class Conflict(ClientError):
    """HTTP 409 Conflict."""

class AlreadyExists(Conflict):
    """gRPC ALREADY_EXISTS error."""

class Aborted(Conflict):
    """gRPC ABORTED error."""

class LengthRequired(ClientError):
    """HTTP 411 Length Required."""

class PreconditionFailed(ClientError):
    """HTTP 412 Precondition Failed."""

class RequestRangeNotSatisfiable(ClientError):
    """HTTP 416 Request Range Not Satisfiable."""

class TooManyRequests(ClientError):
    """HTTP 429 Too Many Requests."""

class ResourceExhausted(TooManyRequests):
    """gRPC RESOURCE_EXHAUSTED error."""

class Cancelled(ClientError):
    """HTTP 499 or gRPC CANCELLED error."""

# 5xx Server Error
class ServerError(GoogleAPICallError):
    """Base class for HTTP 5xx server error responses."""

class InternalServerError(ServerError):
    """HTTP 500 Internal Server Error or gRPC INTERNAL error."""

class Unknown(ServerError):
    """gRPC UNKNOWN error."""

class DataLoss(ServerError):
    """gRPC DATA_LOSS error."""

class MethodNotImplemented(ServerError):
    """HTTP 501 Not Implemented or gRPC UNIMPLEMENTED error."""

class BadGateway(ServerError):
    """HTTP 502 Bad Gateway."""

class ServiceUnavailable(ServerError):
    """HTTP 503 Service Unavailable or gRPC UNAVAILABLE error."""

class GatewayTimeout(ServerError):
    """HTTP 504 Gateway Timeout."""

class DeadlineExceeded(GatewayTimeout):
    """gRPC DEADLINE_EXCEEDED error."""

# Async REST specific
class AsyncRestUnsupportedParameterError(GoogleAPIError):
    """Exception for unsupported parameters in async REST transport."""

Exception Utility Functions

Functions for creating and converting exceptions between different transport protocols.

def exception_class_for_http_status(status_code):
    """
    Return the appropriate exception class for the given HTTP status code.
    
    Args:
        status_code (int): HTTP status code
        
    Returns:
        type: Exception class corresponding to the status code
    """

def from_http_status(status_code, message, **kwargs):
    """
    Create a GoogleAPICallError from an HTTP status code.
    
    Args:
        status_code (int): HTTP status code
        message (str): Error message
        **kwargs: Additional keyword arguments for the exception
        
    Returns:
        GoogleAPICallError: Exception instance for the status code
    """

def format_http_response_error(response, method, url, payload):
    """
    Create an error message and response details from an HTTP response.
    
    Args:
        response: HTTP response object
        method (str): HTTP method used
        url (str): Request URL
        payload: Request payload
        
    Returns:
        str: Formatted error message
    """

def from_http_response(response):
    """
    Create a GoogleAPICallError from a requests.Response object.
    
    Args:
        response: requests.Response object
        
    Returns:
        GoogleAPICallError: Exception created from the response
    """

def exception_class_for_grpc_status(status_code):
    """
    Return the appropriate exception class for the given gRPC status code.
    
    Args:
        status_code (int): gRPC status code
        
    Returns:
        type: Exception class corresponding to the gRPC status
    """

def from_grpc_status(status_code, message, **kwargs):
    """
    Create a GoogleAPICallError from a gRPC status code.
    
    Args:
        status_code (int): gRPC status code
        message (str): Error message
        **kwargs: Additional keyword arguments for the exception
        
    Returns:
        GoogleAPICallError: Exception instance for the status code
    """

def from_grpc_error(rpc_exc):
    """
    Create a GoogleAPICallError from a gRPC RpcError.
    
    Args:
        rpc_exc: gRPC RpcError exception
        
    Returns:
        GoogleAPICallError: Exception created from the gRPC error
    """

Usage Examples

Basic Exception Handling

from google.api_core import exceptions
import requests

try:
    response = requests.get("https://api.example.com/data")
    response.raise_for_status()
except requests.HTTPError as http_error:
    # Convert to Google API Core exception
    api_error = exceptions.from_http_response(response)
    
    if isinstance(api_error, exceptions.NotFound):
        print("Resource not found")
    elif isinstance(api_error, exceptions.ServerError):
        print("Server error occurred")
    else:
        print(f"API error: {api_error}")

Creating Custom Exception Predicates

from google.api_core import exceptions

def is_retryable_error(exception):
    """Check if an exception should trigger a retry."""
    return isinstance(exception, (
        exceptions.InternalServerError,
        exceptions.ServiceUnavailable,
        exceptions.DeadlineExceeded,
        exceptions.TooManyRequests
    ))

# Use with retry logic
from google.api_core import retry
retry_config = retry.Retry(predicate=is_retryable_error)

Exception Status Code Mapping

from google.api_core import exceptions

# Map HTTP status codes to exceptions
error_class = exceptions.exception_class_for_http_status(404)
print(error_class)  # <class 'google.api_core.exceptions.NotFound'>

# Create exception from status
error = exceptions.from_http_status(500, "Internal server error")
print(type(error))  # <class 'google.api_core.exceptions.InternalServerError'>

Install with Tessl CLI