Skip to main content
Proper error handling ensures your applications are robust and provide clear feedback when issues occur.

​
SDK Exceptions

​
Common Exception Types

The SDK provides specific exception types: 
from infrahub_sdk.exceptions import (
    GraphQLError,          # GraphQL API errors
    NodeNotFoundError,     # Object not found
    ValidationError,       # Validation failures
    AuthenticationError,   # Authentication issues
    ServerError           # Server-side errors
)

​
GraphQLError

Most common error when operations fail: 
from infrahub_sdk import InfrahubClient
from infrahub_sdk.exceptions import GraphQLError

client = InfrahubClient()

try:
    device = await client.create(
        kind="InfraDevice",
        name="",  # Invalid: empty name
        serial_number="SN123456"
    )
    await device.save()
    
except GraphQLError as e:
    print(f"GraphQL error: {e.message}")
    print(f"Query: {e.query}")
    print(f"Variables: {e.variables}")

​
NodeNotFoundError

When querying non-existent objects: 
from infrahub_sdk.exceptions import NodeNotFoundError

try:
    device = await client.get(
        kind="InfraDevice",
        id="nonexistent-id"
    )
except NodeNotFoundError:
    print("Device not found")
except GraphQLError as e:
    print(f"Other error: {e.message}")

​
ValidationError

Validation failures: 
from infrahub_sdk.exceptions import ValidationError

try:
    device = await client.create(
        kind="InfraDevice",
        name="invalid@name",  # Invalid characters
        serial_number="SN123456"
    )
    await device.save()
    
except ValidationError as e:
    print(f"Validation failed: {e.message}")
    print(f"Field: {e.field}")

​
Basic Error Handling

​
Try-Except Pattern

Standard error handling: 
import asyncio
from infrahub_sdk import InfrahubClient
from infrahub_sdk.exceptions import GraphQLError

async def create_device_safely():
    client = InfrahubClient()
    
    try:
        device = await client.create(
            kind="InfraDevice",
            name="router-01",
            serial_number="SN123456"
        )
        await device.save()
        print(f"Created device: {device.id}")
        return device
        
    except GraphQLError as e:
        print(f"Failed to create device: {e.message}")
        return None
    except Exception as e:
        print(f"Unexpected error: {e}")
        return None

if __name__ == "__main__":
    asyncio.run(create_device_safely())

​
Multiple Exception Types

Handle different errors appropriately: 
from infrahub_sdk.exceptions import (
    GraphQLError,
    NodeNotFoundError,
    ValidationError
)

try:
    device = await client.get(
        kind="InfraDevice",
        id="device-id"
    )
    device.name.value = "new-name"
    await device.save()
    
except NodeNotFoundError:
    print("Device not found")
except ValidationError as e:
    print(f"Invalid data: {e.message}")
except GraphQLError as e:
    print(f"GraphQL error: {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

​
Error Recovery

​
Retry on Failure

Automatically retry failed operations: 
import asyncio
from infrahub_sdk.exceptions import GraphQLError

async def create_with_retry(
    client: InfrahubClient,
    max_retries: int = 3
):
    """Create device with automatic retry."""
    for attempt in range(max_retries):
        try:
            device = await client.create(
                kind="InfraDevice",
                name="router-01",
                serial_number="SN123456"
            )
            await device.save()
            print(f"Success on attempt {attempt + 1}")
            return device
            
        except GraphQLError as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Attempt {attempt + 1} failed, retrying in {wait_time}s...")
                await asyncio.sleep(wait_time)
            else:
                print(f"Failed after {max_retries} attempts")
                raise

​
Fallback Values

Provide defaults when operations fail: 
async def get_device_or_default(
    client: InfrahubClient,
    device_id: str
):
    """Get device or return default values."""
    try:
        return await client.get(
            kind="InfraDevice",
            id=device_id
        )
    except NodeNotFoundError:
        print(f"Device {device_id} not found, using defaults")
        return {
            "id": device_id,
            "name": "Unknown Device",
            "is_active": False
        }

​
Partial Failure Handling

Continue processing when some operations fail: 
async def batch_create_with_error_handling(
    client: InfrahubClient,
    devices_data: list[dict]
):
    """Create devices, tracking successes and failures."""
    results = {
        "success": [],
        "failed": []
    }
    
    for data in devices_data:
        try:
            device = await client.create(
                kind="InfraDevice",
                **data
            )
            await device.save()
            results["success"].append(device)
        
        except GraphQLError as e:
            results["failed"].append({
                "data": data,
                "error": str(e)
            })
    
    print(f"Created: {len(results['success'])}")
    print(f"Failed: {len(results['failed'])}")
    
    return results

​
Validation and Constraints

​
Pre-Validation

Validate data before sending to API: 
import re

def validate_device_data(data: dict) -> tuple[bool, str]:
    """Validate device data before creating."""
    # Check required fields
    if not data.get("name"):
        return False, "Name is required"
    
    # Validate name format
    if not re.match(r'^[a-z0-9-]+$', data["name"]):
        return False, "Name must be lowercase alphanumeric with hyphens"
    
    # Validate serial number
    serial = data.get("serial_number", "")
    if not serial.startswith("SN"):
        return False, "Serial number must start with 'SN'"
    
    return True, "Valid"

# Usage
device_data = {
    "name": "router-01",
    "serial_number": "SN123456"
}

is_valid, message = validate_device_data(device_data)

if is_valid:
    device = await client.create(kind="InfraDevice", **device_data)
    await device.save()
else:
    print(f"Validation failed: {message}")

​
Handle Constraint Violations

from infrahub_sdk.exceptions import GraphQLError

try:
    device = await client.create(
        kind="InfraDevice",
        name="router-01",
        serial_number="duplicate-serial"
    )
    await device.save()
    
except GraphQLError as e:
    if "unique" in e.message.lower():
        print("Constraint violation: Serial number must be unique")
        # Handle duplicate
    elif "required" in e.message.lower():
        print("Missing required field")
        # Handle missing data
    else:
        print(f"Other error: {e.message}")

​
Connection Errors

​
Handle Connection Failures

import httpx
from infrahub_sdk import InfrahubClient

async def check_connection():
    """Verify connection to Infrahub."""
    client = InfrahubClient()
    
    try:
        # Test connection with simple query
        await client.schema.fetch()
        print("✓ Connected to Infrahub")
        return True
        
    except httpx.ConnectError:
        print("✗ Connection refused - is Infrahub running?")
        return False
    except httpx.TimeoutException:
        print("✗ Connection timeout - check network")
        return False
    except Exception as e:
        print(f"✗ Connection error: {e}")
        return False

​
Timeout Configuration

from infrahub_sdk import Config, InfrahubClient

# Configure longer timeout
config = Config(
    address="https://infrahub.example.com",
    timeout=60  # 60 seconds
)

client = InfrahubClient(config=config)

try:
    # Long-running operation
    devices = await client.all(kind="InfraDevice")
except httpx.TimeoutException:
    print("Operation timed out")

​
Authentication Errors

​
Handle Auth Failures

from infrahub_sdk import Config, InfrahubClient
from infrahub_sdk.exceptions import AuthenticationError

async def authenticate():
    """Test authentication."""
    config = Config(
        address="https://infrahub.example.com",
        api_token="your-token"
    )
    
    client = InfrahubClient(config=config)
    
    try:
        await client.schema.fetch()
        print("✓ Authentication successful")
        return True
        
    except AuthenticationError:
        print("✗ Invalid API token")
        return False
    except Exception as e:
        print(f"✗ Auth error: {e}")
        return False

​
Error Logging

​
Basic Logging

import logging
from infrahub_sdk.exceptions import GraphQLError

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

async def create_device_with_logging(client: InfrahubClient):
    try:
        device = await client.create(
            kind="InfraDevice",
            name="router-01",
            serial_number="SN123456"
        )
        await device.save()
        logger.info(f"Created device: {device.id}")
        return device
        
    except GraphQLError as e:
        logger.error(f"Failed to create device: {e.message}")
        logger.debug(f"Query: {e.query}")
        logger.debug(f"Variables: {e.variables}")
        raise

​
Structured Logging

import structlog
from infrahub_sdk.exceptions import GraphQLError

logger = structlog.get_logger()

async def create_device_structured_logging(client: InfrahubClient):
    try:
        device = await client.create(
            kind="InfraDevice",
            name="router-01",
            serial_number="SN123456"
        )
        await device.save()
        
        logger.info(
            "device_created",
            device_id=device.id,
            device_name=device.name.value
        )
        return device
        
    except GraphQLError as e:
        logger.error(
            "device_creation_failed",
            error_message=e.message,
            device_name="router-01"
        )
        raise

​
Error Context

​
Provide Detailed Error Information

class DeviceCreationError(Exception):
    """Custom exception for device creation failures."""
    
    def __init__(self, device_data: dict, original_error: Exception):
        self.device_data = device_data
        self.original_error = original_error
        
        message = f"Failed to create device {device_data.get('name')}: {original_error}"
        super().__init__(message)

async def create_device_with_context(client: InfrahubClient, device_data: dict):
    try:
        device = await client.create(
            kind="InfraDevice",
            **device_data
        )
        await device.save()
        return device
        
    except GraphQLError as e:
        raise DeviceCreationError(device_data, e)

# Usage
try:
    device = await create_device_with_context(
        client,
        {"name": "router-01", "serial_number": "SN123456"}
    )
except DeviceCreationError as e:
    print(f"Device creation failed")
    print(f"Data: {e.device_data}")
    print(f"Error: {e.original_error}")

​
Graceful Degradation

​
Fallback Behavior

async def get_devices_with_fallback(client: InfrahubClient):
    """Get devices with fallback to cache."""
    try:
        # Try to get fresh data
        devices = await client.all(kind="InfraDevice")
        # Cache the results
        cache_devices(devices)
        return devices
        
    except Exception as e:
        print(f"Error fetching devices: {e}")
        # Fall back to cached data
        cached = get_cached_devices()
        if cached:
            print("Using cached device data")
            return cached
        else:
            print("No cached data available")
            return []

def cache_devices(devices):
    # Implementation
    pass

def get_cached_devices():
    # Implementation
    pass

​
Testing Error Scenarios

​
Mock Errors for Testing

import pytest
from unittest.mock import AsyncMock, patch
from infrahub_sdk.exceptions import GraphQLError

@pytest.mark.asyncio
async def test_error_handling():
    """Test device creation error handling."""
    client = AsyncMock()
    
    # Mock a GraphQL error
    client.create.side_effect = GraphQLError(
        message="Validation failed",
        query="...",
        variables={}
    )
    
    with pytest.raises(GraphQLError):
        await create_device_safely(client)

​
Best Practices

Catch specific exceptions before general ones to provide targeted error handling.
Include relevant context (IDs, names, operations) in error logs for debugging.
Validate data before sending to the API to catch errors early.
Use exponential backoff for transient errors.
Convert technical errors into user-friendly messages.
Use try-finally or context managers to ensure cleanup happens even on errors.

​
Next Steps

Async Operations

Handle errors in async contexts

Batch Operations

Error handling in batch operations

Pagination

Handle pagination errors

Client Setup

Configure error handling behavior

Build docs developers (and LLMs) love

Get started for freeTalk to us