tessl/pypi-msrest

AutoRest swagger generator Python client runtime for REST API clients with serialization, authentication, and request handling.

—

Pending

Overview

Eval results

Files

Exception Handling

Name: tessl/pypi-msrest
Author: tessl

Comprehensive error handling with specialized exception types for validation errors, HTTP operation failures, authentication issues, and client request problems. The exception system provides detailed error information and supports proper error handling patterns.

Capabilities

Base Exception

Base exception class for all MSRest client runtime errors.

class ClientException(Exception):
    def __init__(self, message: str, inner_exception=None, *args, **kwargs):
        """
        Base client exception.
        
        Parameters:
        - message: Error description
        - inner_exception: Nested exception (optional)
        - args: Additional positional arguments
        - kwargs: Additional keyword arguments
        """
    
    inner_exception: any

Validation Errors

Errors that occur during request parameter validation.

class ValidationError(ClientException):
    def __init__(self, rule: str, target: str, value: str, *args, **kwargs):
        """
        Request parameter validation error.
        
        Parameters:
        - rule: Validation rule that failed
        - target: Parameter name that failed validation
        - value: Invalid value that caused the error
        - args: Additional arguments
        - kwargs: Additional keyword arguments
        """
    
    rule: str  # Validation rule (e.g., 'required', 'min_length')
    target: str  # Parameter name
    
    # Validation rule messages
    _messages: dict = {
        "min_length": "must have length greater than {!r}.",
        "max_length": "must have length less than {!r}.",
        "minimum_ex": "must be greater than {!r}.",
        "maximum_ex": "must be less than {!r}.",
        "minimum": "must be equal to or greater than {!r}.",
        "maximum": "must be equal to or less than {!r}.",
        "min_items": "must contain at least {!r} items.",
        "max_items": "must contain at most {!r} items.",
        "pattern": "must conform to the following pattern: {!r}.",
        "unique": "must contain only unique items.",
        "multiple": "must be a multiple of {!r}.",
        "required": "can not be None.",
        "type": "must be of type {!r}"
    }

HTTP Operation Errors

Errors that occur during HTTP operations with detailed response information.

class HttpOperationError(ClientException):
    def __init__(self, deserialize, response, resp_type=None, *args, **kwargs):
        """
        HTTP operation error with response details.
        
        Parameters:
        - deserialize: Deserializer for error response
        - response: HTTP response object
        - resp_type: Expected response type for deserialization
        - args: Additional arguments
        - kwargs: Additional keyword arguments
        """
    
    error: any  # Deserialized error object
    message: str  # Error message
    response: any  # HTTP response object
    
    _DEFAULT_MESSAGE: str = "Unknown error"

Authentication Errors

Errors related to authentication and authorization.

class AuthenticationError(ClientException):
    """Client request failed to authenticate."""

class TokenExpiredError(ClientException):
    """OAuth token expired, request failed."""

Client Request Errors

General client request failures.

class ClientRequestError(ClientException):
    """Client request failed."""

Serialization Errors

Errors during data serialization and deserialization (imported from azure.core).

# Imported from azure.core.exceptions
class SerializationError(Exception):
    """Error during data serialization."""

class DeserializationError(Exception):
    """Error during data deserialization."""

Utility Functions

Helper functions for exception handling.

def raise_with_traceback(exception, message="", *args, **kwargs):
    """
    Raise exception preserving original traceback.
    Must be called within an except block.
    
    Parameters:
    - exception: Exception class to raise
    - message: Error message
    - args: Additional arguments for exception
    - kwargs: Additional keyword arguments for exception
    """

Usage Examples

Basic Exception Handling

from msrest import ServiceClient, Configuration
from msrest.exceptions import (
    ClientException, ValidationError, HttpOperationError,
    AuthenticationError, ClientRequestError
)

config = Configuration(base_url='https://api.example.com')
client = ServiceClient(None, config)

try:
    request = client.get('/data')
    response = client.send(request)
    
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
    # Handle authentication error (e.g., refresh token)
    
except HttpOperationError as e:
    print(f"HTTP operation failed: {e}")
    print(f"Status code: {e.response.status_code}")
    print(f"Response body: {e.response.text}")
    
except ClientRequestError as e:
    print(f"Client request error: {e}")
    
except ClientException as e:
    print(f"General client error: {e}")
    if e.inner_exception:
        print(f"Inner exception: {e.inner_exception}")

Validation Error Handling

from msrest.serialization import Serializer
from msrest.exceptions import ValidationError

serializer = Serializer()

try:
    # This will fail validation
    result = serializer.url('user_id', None, 'int', required=True)
    
except ValidationError as e:
    print(f"Validation failed for parameter '{e.target}'")
    print(f"Rule: {e.rule}")
    print(f"Message: {e}")
    
    # Handle specific validation rules
    if e.rule == 'required':
        print("Parameter is required but was None")
    elif e.rule == 'min_length':
        print("Value is too short")
    elif e.rule == 'pattern':
        print("Value doesn't match required pattern")

HTTP Error Response Handling

from msrest.exceptions import HttpOperationError
from msrest.serialization import Deserializer
import json

def handle_api_errors(client, request):
    """Handle various HTTP error responses."""
    
    try:
        response = client.send(request)
        return response
        
    except HttpOperationError as e:
        status_code = e.response.status_code
        
        # Handle specific status codes
        if status_code == 400:
            print("Bad Request - check request parameters")
            # Try to parse error details
            try:
                error_data = json.loads(e.response.text)
                print(f"Error details: {error_data}")
            except json.JSONDecodeError:
                print(f"Raw error response: {e.response.text}")
                
        elif status_code == 401:
            print("Unauthorized - authentication required")
            raise AuthenticationError("Authentication required")
            
        elif status_code == 403:
            print("Forbidden - insufficient permissions")
            
        elif status_code == 404:
            print("Not Found - resource doesn't exist")
            
        elif status_code == 429:
            print("Rate Limited - too many requests")
            # Could implement retry with backoff
            
        elif status_code >= 500:
            print(f"Server Error ({status_code}) - service unavailable")
            # Could implement retry logic
            
        else:
            print(f"Unexpected HTTP error: {status_code}")
        
        # Re-raise the original exception
        raise

# Usage
try:
    response = handle_api_errors(client, request)
except Exception as e:
    print(f"Request ultimately failed: {e}")

Custom Exception Classes

from msrest.exceptions import ClientException

class CustomAPIError(ClientException):
    """Custom exception for specific API errors."""
    
    def __init__(self, error_code, error_message, response=None):
        super(CustomAPIError, self).__init__(error_message)
        self.error_code = error_code
        self.response = response

class RateLimitError(CustomAPIError):
    """Rate limiting error with retry information."""
    
    def __init__(self, retry_after=None, response=None):
        message = f"Rate limit exceeded"
        if retry_after:
            message += f", retry after {retry_after} seconds"
        
        super(RateLimitError, self).__init__("RATE_LIMIT", message, response)
        self.retry_after = retry_after

# Usage with custom exceptions
def make_api_request(client, request):
    try:
        response = client.send(request)
        return response
        
    except HttpOperationError as e:
        if e.response.status_code == 429:
            # Extract retry-after header
            retry_after = e.response.headers.get('Retry-After')
            if retry_after:
                retry_after = int(retry_after)
            raise RateLimitError(retry_after, e.response)
        
        # Check for custom API error format
        try:
            error_data = json.loads(e.response.text)
            if 'error_code' in error_data:
                raise CustomAPIError(
                    error_data['error_code'],
                    error_data.get('error_message', 'Unknown error'),
                    e.response
                )
        except (json.JSONDecodeError, KeyError):
            pass
        
        # Re-raise original exception if not handled
        raise

# Handle custom exceptions
try:
    response = make_api_request(client, request)
except RateLimitError as e:
    print(f"Rate limited: {e}")
    if e.retry_after:
        print(f"Can retry after {e.retry_after} seconds")
except CustomAPIError as e:
    print(f"API Error {e.error_code}: {e}")

Serialization Error Handling

from msrest.serialization import Serializer, Deserializer, Model
from msrest.exceptions import SerializationError, DeserializationError

class User(Model):
    _attribute_map = {
        'name': {'key': 'name', 'type': 'str'},
        'age': {'key': 'age', 'type': 'int'}
    }

serializer = Serializer({'User': User})
deserializer = Deserializer({'User': User})

# Serialization error handling
try:
    # This might fail if data is invalid
    user_data = {'name': 'John', 'age': 'invalid_age'}
    serialized = serializer.body(user_data, 'User')
    
except SerializationError as e:
    print(f"Serialization failed: {e}")
    # Could try data type conversion or validation

# Deserialization error handling
try:
    # This might fail with invalid JSON or missing fields
    response_text = '{"name": "John", "age": "not_a_number"}'
    user = deserializer('User', response_text)
    
except DeserializationError as e:
    print(f"Deserialization failed: {e}")
    # Could implement fallback parsing or error recovery

Exception Context and Traceback

import sys
import traceback
from msrest.exceptions import raise_with_traceback, ClientException

def process_with_context():
    """Demonstrate exception handling with context preservation."""
    
    try:
        # Some operation that might fail
        result = risky_operation()
        return result
        
    except ValueError as e:
        # Preserve original traceback while raising different exception
        try:
            raise_with_traceback(
                ClientException,
                f"Processing failed: {e}",
                inner_exception=e
            )
        except Exception:
            # If raise_with_traceback fails, raise normally
            raise ClientException(f"Processing failed: {e}", inner_exception=e)
    
    except Exception as e:
        # Log full traceback for debugging
        print("Unexpected error occurred:")
        traceback.print_exc()
        
        # Create exception with context
        exc_type, exc_value, exc_traceback = sys.exc_info()
        raise ClientException(
            f"Unexpected error: {exc_type.__name__}: {exc_value}",
            inner_exception=e
        )

# Usage
try:
    result = process_with_context()
except ClientException as e:
    print(f"Process failed: {e}")
    if e.inner_exception:
        print(f"Root cause: {type(e.inner_exception).__name__}: {e.inner_exception}")

Comprehensive Error Handling Strategy

import time
import logging
from msrest.exceptions import *

# Configure logging for error tracking
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class APIClient:
    """Client with comprehensive error handling."""
    
    def __init__(self, client):
        self.client = client
        self.max_retries = 3
        self.base_retry_delay = 1
    
    def make_request(self, request, **kwargs):
        """Make request with full error handling and retry logic."""
        
        last_exception = None
        
        for attempt in range(self.max_retries + 1):
            try:
                response = self.client.send(request, **kwargs)
                
                # Success
                logger.info(f"Request successful on attempt {attempt + 1}")
                return response
                
            except ValidationError as e:
                # Validation errors are not retryable
                logger.error(f"Validation error: {e}")
                raise
                
            except AuthenticationError as e:
                # Authentication errors might be retryable if token can be refreshed
                logger.warning(f"Authentication error: {e}")
                if attempt < self.max_retries:
                    logger.info("Attempting to refresh authentication")
                    # Implement token refresh logic here
                    time.sleep(1)
                    continue
                raise
                
            except HttpOperationError as e:
                status_code = e.response.status_code
                
                # Client errors (4xx) - generally not retryable
                if 400 <= status_code < 500:
                    if status_code == 429:  # Rate limit - retryable
                        retry_after = e.response.headers.get('Retry-After', self.base_retry_delay)
                        if attempt < self.max_retries:
                            logger.warning(f"Rate limited, retrying after {retry_after}s")
                            time.sleep(int(retry_after))
                            continue
                    
                    logger.error(f"Client error ({status_code}): {e}")
                    raise
                
                # Server errors (5xx) - retryable
                elif status_code >= 500:
                    if attempt < self.max_retries:
                        delay = self.base_retry_delay * (2 ** attempt)  # Exponential backoff
                        logger.warning(f"Server error ({status_code}), retrying in {delay}s")
                        time.sleep(delay)
                        last_exception = e
                        continue
                    
                    logger.error(f"Server error after {self.max_retries} retries: {e}")
                    raise
                
                else:
                    logger.error(f"Unexpected HTTP status {status_code}: {e}")
                    raise
                    
            except ClientRequestError as e:
                # Network/connection errors - retryable
                if attempt < self.max_retries:
                    delay = self.base_retry_delay * (2 ** attempt)
                    logger.warning(f"Request error, retrying in {delay}s: {e}")
                    time.sleep(delay)
                    last_exception = e
                    continue
                
                logger.error(f"Request failed after {self.max_retries} retries: {e}")
                raise
                
            except ClientException as e:
                # General client exceptions
                logger.error(f"Client exception: {e}")
                if e.inner_exception:
                    logger.error(f"Inner exception: {e.inner_exception}")
                raise
                
            except Exception as e:
                # Unexpected errors
                logger.error(f"Unexpected error: {type(e).__name__}: {e}")
                raise ClientException(f"Unexpected error: {e}", inner_exception=e)
        
        # All retries exhausted
        if last_exception:
            raise last_exception

# Usage
api_client = APIClient(client)

try:
    request = client.get('/api/data')
    response = api_client.make_request(request)
    print("Request successful")
    
except ValidationError as e:
    print(f"Invalid request parameters: {e}")
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
except HttpOperationError as e:
    print(f"HTTP operation failed: {e}")
except ClientException as e:
    print(f"Client error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Exception Hierarchy

Exception
└── ClientException
    ├── ValidationError
    ├── ClientRequestError
    ├── AuthenticationError
    │   └── TokenExpiredError
    └── HttpOperationError

Error Response Formats

Common error response formats that HttpOperationError can deserialize:

# OData v4 format (used by Azure ARM)
{
    "error": {
        "code": "ResourceNotFound",
        "message": "The requested resource was not found"
    }
}

# Simple format
{
    "message": "Validation failed for field 'email'"
}

# Detailed format
{
    "error_code": "INVALID_INPUT",
    "error_message": "Email address is required",
    "details": {
        "field": "email",
        "value": null
    }
}

Install with Tessl CLI