tessl/pypi-azure-core

Microsoft Azure Core Library providing foundational infrastructure for Azure SDK Python clients

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Error Handling and Exceptions

Name: tessl/pypi-azure-core
Author: tessl

Azure Core provides a comprehensive exception hierarchy designed to handle various error conditions that can occur when interacting with Azure services. The exception system enables consistent error handling across all Azure SDK clients.

Core Exception Hierarchy

All Azure Core exceptions inherit from the base AzureError class, providing a consistent interface for error handling.

from azure.core.exceptions import AzureError

class AzureError(Exception):
    """Base exception for all Azure SDK errors."""
    def __init__(self, message: str = None, **kwargs): ...
    
    @property
    def message(self) -> str: ...
    
    @property
    def error(self) -> Optional[object]: ...
    
    @property
    def continuation_token(self) -> Optional[str]: ...

Capabilities

Service Request and Response Errors

Errors that occur during HTTP request preparation and response processing.

from azure.core.exceptions import ServiceRequestError, ServiceResponseError, HttpResponseError

class ServiceRequestError(AzureError):
    """Error occurred during request preparation."""
    def __init__(self, message: str = None, **kwargs): ...

class ServiceResponseError(AzureError):
    """Error occurred during response processing."""
    def __init__(self, message: str = None, **kwargs): ...

class HttpResponseError(AzureError):
    """HTTP response error with status code information."""
    def __init__(self, message: str = None, response: HttpResponse = None, **kwargs): ...
    
    @property
    def status_code(self) -> Optional[int]: ...
    
    @property
    def response(self) -> Optional[HttpResponse]: ...
    
    @property
    def reason(self) -> Optional[str]: ...

HTTP Status Code Specific Errors

Specialized exceptions for common HTTP status codes and error conditions.

from azure.core.exceptions import (
    ResourceExistsError, ResourceNotFoundError, ResourceModifiedError,
    ResourceNotModifiedError, ClientAuthenticationError
)

class ResourceExistsError(HttpResponseError):
    """Resource already exists error (typically 409)."""
    pass

class ResourceNotFoundError(HttpResponseError):
    """Resource not found error (404/412)."""
    pass

class ResourceModifiedError(HttpResponseError):
    """Resource was modified error (412 with etag mismatch)."""
    pass

class ResourceNotModifiedError(HttpResponseError):
    """Resource not modified error (304).""" 
    pass

class ClientAuthenticationError(HttpResponseError):
    """Authentication error (401/403)."""
    pass

Network and Transport Errors

Errors related to network connectivity and HTTP transport.

from azure.core.exceptions import (
    TooManyRedirectsError, ServiceRequestTimeoutError, ServiceResponseTimeoutError,
    IncompleteReadError
)

class TooManyRedirectsError(ServiceRequestError):
    """Maximum redirect attempts exceeded."""
    def __init__(self, history: Optional[List] = None, **kwargs): ...
    
    @property
    def history(self) -> Optional[List]: ...

class ServiceRequestTimeoutError(ServiceRequestError):
    """Service request timeout error."""
    pass

class ServiceResponseTimeoutError(ServiceResponseError):
    """Service response timeout error."""
    pass

class IncompleteReadError(ServiceResponseError):
    """Incomplete read error during response processing."""
    pass

Stream and Content Errors

Errors related to request/response stream handling and content processing.

from azure.core.exceptions import (
    StreamConsumedError, StreamClosedError, ResponseNotReadError
)

class StreamConsumedError(AzureError):
    """Stream has already been consumed and cannot be read again."""
    pass

class StreamClosedError(AzureError):
    """Stream has been closed and cannot be accessed."""
    pass

class ResponseNotReadError(AzureError):
    """Response content has not been read yet."""
    pass

Serialization and Deserialization Errors

Errors that occur during data serialization and deserialization.

from azure.core.exceptions import SerializationError, DeserializationError, DecodeError

class SerializationError(AzureError):
    """Error occurred during data serialization."""
    pass

class DeserializationError(AzureError):
    """Error occurred during data deserialization."""
    pass

class DecodeError(ServiceResponseError):
    """Response content decoding error."""
    pass

OData Error Handling

Support for OData V4 error format parsing commonly used by Azure services.

from azure.core.exceptions import ODataV4Format, ODataV4Error

class ODataV4Format:
    """OData V4 error format parser."""
    @staticmethod
    def deserialize(response: HttpResponse) -> Dict: ...

class ODataV4Error(HttpResponseError):
    """OData V4 formatted error response."""
    def __init__(self, response: HttpResponse, **kwargs): ...
    
    @property
    def code(self) -> Optional[str]: ...
    
    @property
    def details(self) -> Optional[List[Dict]]: ...

Exception Mapping Utilities

Utilities for mapping HTTP status codes to appropriate exception types.

from azure.core.exceptions import map_error
from typing import Dict, Type, Callable

def map_error(
    status_code: int,
    response: HttpResponse,
    error_map: Optional[Dict[int, Type[HttpResponseError]]] = None
) -> None:
    """Map HTTP status codes to exception types and raise appropriate error."""
    ...

# Deprecated utility function
def raise_with_traceback(exception: Exception, traceback=None) -> None:
    """Raise exception with traceback (deprecated)."""
    ...

Usage Examples

Basic Exception Handling

from azure.core.exceptions import AzureError, HttpResponseError, ResourceNotFoundError
from azure.core import PipelineClient

try:
    client = PipelineClient(base_url="https://api.example.com")
    # Make request...
except ResourceNotFoundError as e:
    print(f"Resource not found: {e.message}")
    print(f"Status code: {e.status_code}")
except HttpResponseError as e:
    print(f"HTTP error: {e.status_code} - {e.reason}")
except AzureError as e:
    print(f"Azure error: {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

Custom Error Mapping

from azure.core.exceptions import map_error, HttpResponseError

# Define custom error mapping
error_map = {
    404: ResourceNotFoundError,
    409: ResourceExistsError,
    401: ClientAuthenticationError,
}

def handle_response(response):
    if response.status_code >= 400:
        # Use custom error mapping
        map_error(response.status_code, response, error_map)
    return response

Handling Specific Service Errors

from azure.core.exceptions import HttpResponseError, ODataV4Error
import json

try:
    # Make service request...
    pass
except HttpResponseError as e:
    if e.response and e.response.headers.get("content-type", "").startswith("application/json"):
        try:
            error_details = e.response.json()
            if "error" in error_details:
                # Handle OData error format
                error_code = error_details["error"].get("code")
                error_message = error_details["error"].get("message")
                print(f"Service error {error_code}: {error_message}")
            else:
                print(f"Service returned error: {error_details}")
        except (json.JSONDecodeError, KeyError):
            print(f"Non-JSON error response: {e.message}")
    else:
        print(f"HTTP error {e.status_code}: {e.reason}")

Retry Logic with Exception Handling

from azure.core.exceptions import AzureError, ServiceRequestError
import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=1.0):
    """Retry function with exponential backoff."""
    for attempt in range(max_retries + 1):
        try:
            return func()
        except (ServiceRequestError, ConnectionError) as e:
            if attempt == max_retries:
                raise e
            
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"Attempt {attempt + 1} failed, retrying in {delay:.2f}s...")
            time.sleep(delay)
        except AzureError:
            # Don't retry on non-transient Azure errors
            raise

# Usage
def make_request():
    # Your request logic here
    pass

try:
    result = retry_with_backoff(make_request, max_retries=5)
except AzureError as e:
    print(f"Request failed after all retries: {e}")

Stream Error Handling

from azure.core.exceptions import StreamConsumedError, StreamClosedError

def read_stream_safely(response):
    try:
        # Read stream content
        content = response.content
        return content
    except StreamConsumedError:
        print("Stream has already been consumed")
        return None
    except StreamClosedError:
        print("Stream has been closed")
        return None

def read_stream_multiple_times(response):
    """Example of handling stream consumption."""
    try:
        # First read
        content1 = response.text
        
        # Second read will fail
        content2 = response.text
        
    except StreamConsumedError:
        print("Cannot read stream multiple times without reopening")
        # Would need to make a new request to get content again

Async Exception Handling

import asyncio
from azure.core.exceptions import AzureError, HttpResponseError

async def async_request_with_error_handling():
    try:
        # Make async request...
        pass
    except HttpResponseError as e:
        print(f"Async HTTP error: {e.status_code}")
        # Handle specific async error conditions
        if e.status_code == 429:  # Rate limited
            # Wait before retrying
            await asyncio.sleep(1)
    except AzureError as e:
        print(f"Async Azure error: {e.message}")

# asyncio.run(async_request_with_error_handling())

Error Response Processing

OData V4 Error Format

Many Azure services return errors in OData V4 format:

{
  "error": {
    "code": "InvalidResourceName",
    "message": "The specified resource name contains invalid characters.",
    "details": [
      {
        "code": "InvalidCharacter",
        "message": "Character '@' is not allowed in resource names.",
        "target": "resourceName"
      }
    ]
  }
}

from azure.core.exceptions import ODataV4Format

def parse_odata_error(response):
    if response.headers.get("content-type", "").startswith("application/json"):
        error_data = ODataV4Format.deserialize(response)
        if error_data:
            main_error = error_data.get("error", {})
            code = main_error.get("code")
            message = main_error.get("message")
            details = main_error.get("details", [])
            
            print(f"Error code: {code}")
            print(f"Error message: {message}")
            for detail in details:
                print(f"Detail: {detail.get('message')} (target: {detail.get('target')})")

Best Practices

Exception Handling Strategy

Catch specific exceptions before general ones (most specific first)
Always catch AzureError as a fallback for Azure SDK specific errors
Log exception details including status codes and response content when available
Don't ignore exceptions - handle them appropriately or let them bubble up

Error Information Extraction

Use the response property of HttpResponseError to access full response details
Check response headers and content for additional error information
Parse JSON error responses for structured error details
Consider the continuation_token property for resumable operations

Retry and Recovery

Only retry on transient errors (network issues, rate limiting, temporary service errors)
Don't retry on authentication errors or client-side validation errors
Implement exponential backoff with jitter for retry logic
Set reasonable maximum retry limits to avoid infinite loops