Error Handling

Exception Types

When the SDK encounters an error, it raises a specific exception based on the HTTP status code or error type.

Exception Hierarchy

All SDK exceptions inherit from brew_sdk.APIError:

APIError
├── APIConnectionError
├── APIStatusError
│   ├── BadRequestError (400)
│   ├── AuthenticationError (401)
│   ├── PermissionDeniedError (403)
│   ├── NotFoundError (404)
│   ├── ConflictError (409)
│   ├── UnprocessableEntityError (422)
│   ├── RateLimitError (429)
│   └── InternalServerError (>=500)
└── APITimeoutError

Exception Types by Status Code

Status Code	Exception Type	Description
400	`BadRequestError`	Invalid request data
401	`AuthenticationError`	Invalid or missing API key
403	`PermissionDeniedError`	Insufficient permissions
404	`NotFoundError`	Resource not found
409	`ConflictError`	Conflict (e.g., duplicate resource)
422	`UnprocessableEntityError`	Validation error
429	`RateLimitError`	Rate limit exceeded
>=500	`InternalServerError`	Server error
N/A	`APIConnectionError`	Network connectivity issue
N/A	`APITimeoutError`	Request timed out

Handling Errors

Basic Error Handling

import brew_sdk
from brew_sdk import BrewSDK

client = BrewSDK()

try:
    result = client.contacts.import_.create(
        contacts=[{"email": "[email protected]"}],
    )
except brew_sdk.APIError as e:
    print(f"API Error: {e}")

Handling Specific Exception Types

import brew_sdk
from brew_sdk import BrewSDK

client = BrewSDK()

try:
    client.send.transactional.send(
        chat_id="template-id",
        to="[email protected]",
    )
except brew_sdk.BadRequestError as e:
    print(f"Invalid request: {e}")
    # Fix the request data
except brew_sdk.AuthenticationError as e:
    print("Authentication failed - check your API key")
    # Re-authenticate or alert the user
except brew_sdk.NotFoundError as e:
    print("Template not found")
    # Check the template ID
except brew_sdk.RateLimitError as e:
    print("Rate limited - waiting before retry")
    # Wait and retry
except brew_sdk.InternalServerError as e:
    print("Server error - will be retried automatically")
except brew_sdk.APIConnectionError as e:
    print(f"Network error: {e}")
    print(f"Cause: {e.__cause__}")
except brew_sdk.APIStatusError as e:
    # Catch-all for other status errors
    print(f"API error: {e.status_code} - {e.response}")

Exception Properties

APIStatusError and its subclasses include:

try:
    # ...
except brew_sdk.APIStatusError as e:
    print(e.status_code)  # HTTP status code (e.g., 400)
    print(e.response)     # httpx.Response object
    print(e.body)         # Response body (parsed if JSON)
    print(e.message)      # Error message

APIConnectionError includes:

try:
    # ...
except brew_sdk.APIConnectionError as e:
    print(e.__cause__)    # Underlying exception (e.g., httpx error)

Automatic Retries

The SDK automatically retries certain errors with exponential backoff.

Retried Errors

The following errors are automatically retried (2 times by default):

Connection errors (network issues)
408 Request Timeout
409 Conflict
429 Rate Limit
500+ Internal Server Errors

Configure Retry Behavior

from brew_sdk import BrewSDK

# Configure default retries for all requests
client = BrewSDK(max_retries=0)  # Disable retries

# Or configure per-request using with_options
client.with_options(max_retries=5).contacts.import_.create(
    contacts=[{"email": "[email protected]"}],
)

Retry Timing

Retries use exponential backoff:

First retry: ~0.5 seconds
Second retry: ~1 second
Maximum delay: 8 seconds

The SDK also respects Retry-After headers from the server.

Timeouts

Default Timeout

Requests timeout after 60 seconds (1 minute) by default.

Configure Timeouts

import httpx
from brew_sdk import BrewSDK

# Configure default timeout for all requests (in seconds)
client = BrewSDK(timeout=20.0)

# Fine-grained timeout control
client = BrewSDK(
    timeout=httpx.Timeout(
        60.0,        # Total timeout
        read=5.0,    # Read timeout
        write=10.0,  # Write timeout
        connect=2.0, # Connect timeout
    )
)

# Or configure per-request
client.with_options(timeout=5.0).contacts.import_.create(
    contacts=[{"email": "[email protected]"}],
)

Handling Timeouts

import brew_sdk
from brew_sdk import BrewSDK

client = BrewSDK()

try:
    client.send.transactional.send(
        chat_id="template-id",
        to="[email protected]",
    )
except brew_sdk.APITimeoutError:
    print("Request timed out")
    # Note: Timed out requests are retried by default

Logging

The SDK uses Python’s standard logging module.

Enable Logging

Set the BREW_SDK_LOG environment variable:

# Info level
export BREW_SDK_LOG=info

# Debug level (includes HTTP request/response details)
export BREW_SDK_LOG=debug

Programmatic Logging Setup

import logging

# Configure logging for the SDK
logging.basicConfig()
logging.getLogger("brew_sdk").setLevel(logging.DEBUG)

At the debug level, all HTTP requests and responses are logged, including headers and bodies. Some authentication headers are redacted, but sensitive data in request/response bodies may still be visible.

Accessing Raw Responses

Get Response Headers

Use .with_raw_response to access the raw httpx.Response:

from brew_sdk import BrewSDK

client = BrewSDK()

response = client.contacts.import_.with_raw_response.create(
    contacts=[{"email": "[email protected]"}],
)

print(response.headers.get("X-Request-Id"))

# Parse the response data
result = response.parse()
print(result.data.stats)

Async Raw Response

from brew_sdk import AsyncBrewSDK

client = AsyncBrewSDK()

response = await client.contacts.import_.with_raw_response.create(
    contacts=[{"email": "[email protected]"}],
)

print(response.headers)
result = response.parse()

Streaming Responses

For large responses, use .with_streaming_response:

with client.contacts.import_.with_streaming_response.create(
    contacts=[{"email": "[email protected]"}],
) as response:
    print(response.headers.get("X-Request-Id"))
    
    # Read response body lazily
    for line in response.iter_lines():
        print(line)

Distinguishing None Values

In API responses, a field may be explicitly null or missing entirely. Both show as None in Python. Use model_fields_set to distinguish:

result = client.contacts.import_.create(
    contacts=[{"email": "[email protected]"}],
)

if result.data.import_id is None:
    if "import_id" not in result.data.model_fields_set:
        print('Response had no "import_id" key')
    else:
        print('Response had "import_id": null')

Best Practices

Create a Wrapper Function

Create a wrapper for consistent error handling:

from typing import TypeVar, Callable
import brew_sdk

T = TypeVar("T")

def safe_api_call(api_call: Callable[[], T]) -> tuple[T | None, Exception | None]:
    try:
        data = api_call()
        return data, None
    except brew_sdk.RateLimitError as e:
        # Log and potentially retry with backoff
        print("Rate limited, consider reducing request frequency")
        return None, e
    except brew_sdk.AuthenticationError as e:
        # Alert about authentication issues
        print("API key invalid or expired")
        return None, e
    except brew_sdk.APIError as e:
        return None, e

# Usage
result, error = safe_api_call(
    lambda: client.contacts.import_.create(
        contacts=[{"email": "[email protected]"}]
    )
)

if error:
    print(f"Error: {error}")
else:
    print(f"Success: {result}")

Use Context Managers

Always use context managers with async clients:

from brew_sdk import AsyncBrewSDK

async def main():
    async with AsyncBrewSDK() as client:
        try:
            result = await client.contacts.import_.create(
                contacts=[{"email": "[email protected]"}],
            )
        except brew_sdk.APIError as e:
            print(f"Error: {e}")
    # Client is properly closed here

Log Errors with Context

Include context in error logs:

import logging
import brew_sdk
from brew_sdk import BrewSDK

logger = logging.getLogger(__name__)
client = BrewSDK()

def update_contact(email: str, data: dict):
    try:
        return client.contacts.update(email=email, **data)
    except brew_sdk.APIStatusError as e:
        logger.error(
            "Contact update failed",
            extra={
                "email": email,
                "error_type": type(e).__name__,
                "status_code": e.status_code,
                "message": str(e),
            },
        )
        raise

Implement Retry Logic

For critical operations, implement custom retry logic:

import time
import brew_sdk
from brew_sdk import BrewSDK

def send_with_retry(
    client: BrewSDK,
    chat_id: str,
    to: str,
    variables: dict,
    max_attempts: int = 3,
):
    for attempt in range(max_attempts):
        try:
            return client.send.transactional.send(
                chat_id=chat_id,
                to=to,
                variables=variables,
            )
        except brew_sdk.RateLimitError:
            if attempt < max_attempts - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Rate limited, waiting {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise
        except brew_sdk.InternalServerError:
            if attempt < max_attempts - 1:
                print(f"Server error, retrying...")
                time.sleep(1)
            else:
                raise

Production Error Handler

Here’s a production-ready error handler:

from dataclasses import dataclass
from typing import TypeVar, Generic, Callable
import brew_sdk
from brew_sdk import BrewSDK

T = TypeVar("T")


@dataclass
class BrewError:
    type: str
    message: str
    status: int | None
    retryable: bool


@dataclass
class BrewResult(Generic[T]):
    data: T | None = None
    error: BrewError | None = None


def with_brew_error_handling(
    operation: Callable[[], T],
    context: dict | None = None,
) -> BrewResult[T]:
    """Execute a Brew operation with comprehensive error handling."""
    context = context or {}
    
    try:
        data = operation()
        return BrewResult(data=data)
    
    except brew_sdk.RateLimitError as e:
        return BrewResult(error=BrewError(
            type="RateLimitError",
            message="Too many requests. Please slow down.",
            status=429,
            retryable=True,
        ))
    
    except brew_sdk.AuthenticationError as e:
        # Log for ops team - API key may need rotation
        import logging
        logging.error(f"Brew authentication failed: {context}")
        return BrewResult(error=BrewError(
            type="AuthenticationError",
            message="Authentication failed. Please contact support.",
            status=401,
            retryable=False,
        ))
    
    except brew_sdk.APIStatusError as e:
        return BrewResult(error=BrewError(
            type=type(e).__name__,
            message=str(e),
            status=e.status_code,
            retryable=e.status_code >= 500,
        ))


# Usage
client = BrewSDK()

result = with_brew_error_handling(
    lambda: client.contacts.import_.create(contacts=contacts),
    context={"operation": "import_contacts", "count": len(contacts)},
)

if result.error:
    if result.error.retryable:
        # Queue for retry
        pass
    else:
        # Handle permanent failure
        pass
else:
    print(f"Success: {result.data}")

Need Help?

Our team is ready to support you at every step of your journey with Brew. Choose the option that works best for you:

Self-Service Tools
Talk to Our Team

Search Documentation

Type in the “Ask any question” search bar at the top left to instantly find relevant documentation pages.

ChatGPT/Claude Integration

Click “Open in ChatGPT” at the top right of any page to analyze documentation with ChatGPT or Claude for deeper insights.

Email Support

Getting Started

TypeScript SDK

Python SDK

Exception Types

Exception Hierarchy

Exception Types by Status Code

Handling Errors

Basic Error Handling

Handling Specific Exception Types

Exception Properties

Automatic Retries

Retried Errors

Configure Retry Behavior

Retry Timing

Timeouts

Default Timeout

Configure Timeouts

Handling Timeouts

Logging

Enable Logging

Programmatic Logging Setup

Accessing Raw Responses

Get Response Headers

Async Raw Response

Streaming Responses

Distinguishing None Values

Best Practices

Create a Wrapper Function

Use Context Managers

Log Errors with Context

Implement Retry Logic

Production Error Handler

Need Help?

Search Documentation

ChatGPT/Claude Integration

Schedule a Call

Call Us Directly

Slack Channel

Getting Started

TypeScript SDK

Python SDK

​Exception Types

​Exception Hierarchy

​Exception Types by Status Code

​Handling Errors

​Basic Error Handling

​Handling Specific Exception Types

​Exception Properties

​Automatic Retries

​Retried Errors

​Configure Retry Behavior

​Retry Timing

​Timeouts

​Default Timeout

​Configure Timeouts

​Handling Timeouts

​Logging

​Enable Logging

​Programmatic Logging Setup

​Accessing Raw Responses

​Get Response Headers

​Async Raw Response

​Streaming Responses

​Distinguishing None Values

​Best Practices

​Create a Wrapper Function

​Use Context Managers

​Log Errors with Context

​Implement Retry Logic

​Production Error Handler

​Need Help?

Search Documentation

ChatGPT/Claude Integration

Schedule a Call

Call Us Directly

Slack Channel

Email Support

Exception Types

Exception Hierarchy

Exception Types by Status Code

Handling Errors

Basic Error Handling

Handling Specific Exception Types

Exception Properties

Automatic Retries

Retried Errors

Configure Retry Behavior

Retry Timing

Timeouts

Default Timeout

Configure Timeouts

Handling Timeouts

Logging

Enable Logging

Programmatic Logging Setup

Accessing Raw Responses

Get Response Headers

Async Raw Response

Streaming Responses

Distinguishing None Values

Best Practices

Create a Wrapper Function

Use Context Managers

Log Errors with Context

Implement Retry Logic

Production Error Handler

Need Help?