Error Handling

Exception Hierarchy

 SnipeRouteAPIError (base)
├── ValidationError (400/422)
├── AuthenticationError (401)
├── ForbiddenError (403)
├── NotFoundError (404)
├── ConflictError (409)
├── RateLimitError (429)
└── ServerError (500/503)

Exception Classes

SnipeRouteAPIError

Base exception for all API errors.

from sniperoute.exceptions import SnipeRouteAPIError
 
try:
    response = await client.create_intent(intent)
except SnipeRouteAPIError as e:
    print(f"Error: {e.message}")
    print(f"Status: {e.status_code}")
    print(f"Request ID: {e.request_id}")

Attributes:

message (str): Error message
status_code (int): HTTP status code
error_code (str | None): Machine-readable error code
request_id (str | None): Request ID for support

ValidationError

Raised for 400 Bad Request or 422 Unprocessable Entity.

from sniperoute.exceptions import ValidationError
 
try:
    intent = TradeIntentRequest(
        order_type=OrderType.LIMIT  # Missing limit_price!
    )
    response = await client.create_intent(intent)
except ValidationError as e:
    print(f"Validation error: {e.message}")

Common causes:

Missing required fields
Invalid field types
Invalid enum values
Business logic violations (e.g., missing limit_price for limit orders)

AuthenticationError

Raised for 401 Unauthorized.

from sniperoute.exceptions import AuthenticationError
 
try:
    response = await client.create_intent(intent)
except AuthenticationError as e:
    print(f"Auth error: {e.message}")
    # Check API key

Common causes:

Invalid API key
Missing API key
Expired API key
Revoked API key

ForbiddenError

Raised for 403 Forbidden.

from sniperoute.exceptions import ForbiddenError
 
try:
    response = await client.create_intent(intent)
except ForbiddenError as e:
    print(f"Forbidden: {e.message}")
    # Check API key permissions

Common causes:

API key lacks required permissions
Test key used for production resource

NotFoundError

Raised for 404 Not Found.

from sniperoute.exceptions import NotFoundError
 
try:
    intent = await client.get_intent_by_external_id("nonexistent")
except NotFoundError as e:
    print(f"Not found: {e.message}")

Common causes:

Invalid intent_id
Invalid broker connection ID
Resource deleted

ConflictError

Raised for 409 Conflict.

from sniperoute.exceptions import ConflictError
 
try:
    response = await client.create_intent(intent)
except ConflictError as e:
    print(f"Conflict: {e.message}")
    # Use a different intent_id

Common causes:

Duplicate intent_id

RateLimitError

Raised for 429 Too Many Requests.

from sniperoute.exceptions import RateLimitError
 
try:
    response = await client.create_intent(intent)
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
    await asyncio.sleep(e.retry_after)

Attributes:

retry_after (int | None): Seconds to wait before retrying

ServerError

Raised for 500 Internal Server Error or 503 Service Unavailable.

from sniperoute.exceptions import ServerError
 
try:
    response = await client.create_intent(intent)
except ServerError as e:
    print(f"Server error: {e.message}")
    print(f"Request ID: {e.request_id}")
    # Retry with backoff

Error Handling Strategies

Catch Specific Exceptions

from sniperoute.exceptions import (
    ValidationError,
    ConflictError,
    NotFoundError,
    RateLimitError,
    ServerError,
    SnipeRouteAPIError
)
 
try:
    response = await client.create_intent(intent)
 
except ValidationError as e:
    # Fix validation issues
    logger.error(f"Validation error: {e.message}")
 
except ConflictError as e:
    # Generate new intent_id
    logger.warning(f"Duplicate intent_id: {e.message}")
 
except NotFoundError as e:
    # Resource doesn't exist
    logger.error(f"Not found: {e.message}")
 
except RateLimitError as e:
    # Wait and retry
    logger.warning(f"Rate limited. Retry after {e.retry_after}s")
    await asyncio.sleep(e.retry_after)
 
except ServerError as e:
    # Retry with backoff
    logger.error(f"Server error: {e.message} (request_id: {e.request_id})")
 
except SnipeRouteAPIError as e:
    # Catch-all for other errors
    logger.error(f"API error: {e.message}")

Retry with Exponential Backoff

import asyncio
from sniperoute.exceptions import RateLimitError, ServerError
 
async def create_intent_with_retry(
    client,
    intent,
    max_retries=3,
    base_delay=1.0
):
    """Create Trade Intent with exponential backoff."""
 
    for attempt in range(max_retries):
        try:
            return await client.create_intent(intent)
 
        except RateLimitError as e:
            # Use retry_after from response
            wait_time = e.retry_after or (base_delay * (2 ** attempt))
            logger.info(f"Rate limited. Waiting {wait_time}s...")
            await asyncio.sleep(wait_time)
 
        except ServerError as e:
            # Retry server errors with backoff
            if attempt == max_retries - 1:
                raise  # Last attempt failed
 
            wait_time = base_delay * (2 ** attempt)
            logger.info(f"Server error. Retrying in {wait_time}s...")
            await asyncio.sleep(wait_time)
 
    raise Exception(f"Failed after {max_retries} attempts")

Circuit Breaker Pattern

import asyncio
from datetime import datetime, timedelta
from typing import Optional
 
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure_time: Optional[datetime] = None
        self.state = "closed"  # closed, open, half_open
 
    async def call(self, func, *args, **kwargs):
        if self.state == "open":
            # Check if timeout has passed
            if datetime.now() - self.last_failure_time > timedelta(seconds=self.timeout):
                self.state = "half_open"
            else:
                raise Exception("Circuit breaker is open")
 
        try:
            result = await func(*args, **kwargs)
 
            # Success - reset failures
            if self.state == "half_open":
                self.state = "closed"
            self.failures = 0
 
            return result
 
        except ServerError as e:
            self.failures += 1
            self.last_failure_time = datetime.now()
 
            if self.failures >= self.failure_threshold:
                self.state = "open"
                logger.error(f"Circuit breaker opened after {self.failures} failures")
 
            raise
 
# Usage
circuit_breaker = CircuitBreaker()
 
async def create_intent_with_circuit_breaker(client, intent):
    return await circuit_breaker.call(client.create_intent, intent)

Logging Errors

import logging
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
async def create_intent_with_logging(client, intent):
    try:
        response = await client.create_intent(intent)
        logger.info(f"Intent created: {response.intent_id}")
        return response
 
    except SnipeRouteAPIError as e:
        logger.error(
            f"Failed to create intent: {e.message}",
            extra={
                "intent_id": intent.intent_id,
                "status_code": e.status_code,
                "error_code": e.error_code,
                "request_id": e.request_id
            }
        )
        raise

Complete Example

import asyncio
import logging
from sniperoute import SnipeRouteClient
from sniperoute.models import TradeIntentRequest, OrderSide, OrderType
from sniperoute.exceptions import (
    ValidationError,
    ConflictError,
    RateLimitError,
    ServerError,
    SnipeRouteAPIError
)
from decimal import Decimal
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
async def create_intent_with_full_error_handling(
    client: SnipeRouteClient,
    intent: TradeIntentRequest,
    max_retries: int = 3
):
    """Create Trade Intent with comprehensive error handling."""
 
    for attempt in range(max_retries):
        try:
            # Attempt to create intent
            response = await client.create_intent(intent)
            logger.info(f"Intent created: {response.intent_id}")
            return response
 
        except ValidationError as e:
            # Don't retry validation errors
            logger.error(f"Validation error: {e.message}")
            raise
 
        except ConflictError as e:
            # Don't retry conflicts
            logger.error(f"Duplicate intent_id: {e.message}")
            raise
 
        except RateLimitError as e:
            # Wait and retry
            retry_after = e.retry_after or 60
            logger.warning(f"Rate limited. Waiting {retry_after}s...")
            await asyncio.sleep(retry_after)
 
        except ServerError as e:
            # Retry with backoff
            if attempt == max_retries - 1:
                logger.error(f"Server error after {max_retries} attempts")
                logger.error(f"Request ID: {e.request_id}")
                raise
 
            wait_time = 2 ** attempt
            logger.warning(f"Server error. Retrying in {wait_time}s...")
            await asyncio.sleep(wait_time)
 
        except SnipeRouteAPIError as e:
            # Log and re-raise other errors
            logger.error(
                f"API error: {e.message}",
                extra={
                    "status_code": e.status_code,
                    "error_code": e.error_code,
                    "request_id": e.request_id
                }
            )
            raise
 
async def main():
    client = SnipeRouteClient()
 
    intent = TradeIntentRequest(
        intent_id="error_handling_example_001",
        symbol="AAPL",
        side=OrderSide.BUY,
        quantity=Decimal("10"),
        order_type=OrderType.MARKET,
        broker_id="my_broker"
    )
 
    try:
        response = await create_intent_with_full_error_handling(client, intent)
        print(f"Success: {response.intent_id}")
 
    except ValidationError:
        print("Fix validation errors and try again")
 
    except ConflictError:
        print("Use a different intent_id")
 
    except SnipeRouteAPIError as e:
        print(f"API error: {e.message}")
 
if __name__ == "__main__":
    asyncio.run(main())

Next Steps

Error Reference

Complete API error code reference

Error Handling Guide

Error handling best practices

Models