Async Operations - Infrahub

The Infrahub Python SDK is built with async/await for high-performance concurrent operations.

Understanding Async/Await

Why Async?

Async programming allows concurrent I/O operations:

Non-blocking: Don’t wait for API responses
Concurrent: Process multiple requests simultaneously
Efficient: Better resource utilization
Scalable: Handle more operations in less time

Basic Async Pattern

import asyncio
from infrahub_sdk import InfrahubClient

async def main():
    client = InfrahubClient()
    
    # All SDK operations are async
    device = await client.get(kind="InfraDevice", id="device-id")
    print(device.name.value)

if __name__ == "__main__":
    # Run async function
    asyncio.run(main())

Concurrent Operations

Query Multiple Objects Concurrently

Fetch multiple objects at once:

import asyncio
from infrahub_sdk import InfrahubClient

async def get_multiple_devices():
    client = InfrahubClient()
    
    device_ids = ["id-1", "id-2", "id-3", "id-4", "id-5"]
    
    # Create tasks for all queries
    tasks = [
        client.get(kind="InfraDevice", id=device_id)
        for device_id in device_ids
    ]
    
    # Execute all queries concurrently
    devices = await asyncio.gather(*tasks)
    
    print(f"Retrieved {len(devices)} devices concurrently")
    for device in devices:
        print(f"  - {device.name.value}")

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

Create Multiple Objects Concurrently

import asyncio
from infrahub_sdk import InfrahubClient

async def create_device(client: InfrahubClient, index: int):
    """Create a single device."""
    device = await client.create(
        kind="InfraDevice",
        name=f"device-{index:03d}",
        serial_number=f"SN{index:06d}"
    )
    await device.save()
    return device

async def create_devices_concurrently():
    client = InfrahubClient()
    
    # Create 100 devices concurrently
    tasks = [
        create_device(client, i)
        for i in range(100)
    ]
    
    devices = await asyncio.gather(*tasks)
    
    print(f"Created {len(devices)} devices")

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

Update Multiple Objects Concurrently

import asyncio

async def update_device(client: InfrahubClient, device_id: str):
    """Update a single device."""
    device = await client.get(kind="InfraDevice", id=device_id)
    device.is_managed.value = True
    await device.save()
    return device

async def update_all_devices():
    client = InfrahubClient()
    
    # Get all devices
    all_devices = await client.all(kind="InfraDevice")
    
    # Update concurrently
    tasks = [
        update_device(client, device.id)
        for device in all_devices
    ]
    
    updated = await asyncio.gather(*tasks)
    print(f"Updated {len(updated)} devices concurrently")

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

Controlling Concurrency

Using Semaphore for Rate Limiting

Limit the number of concurrent operations:

import asyncio
from infrahub_sdk import InfrahubClient

async def fetch_device_with_limit(
    client: InfrahubClient,
    device_id: str,
    semaphore: asyncio.Semaphore
):
    """Fetch device with concurrency limit."""
    async with semaphore:
        return await client.get(kind="InfraDevice", id=device_id)

async def limited_concurrent_fetch():
    client = InfrahubClient()
    
    # Limit to 10 concurrent requests
    semaphore = asyncio.Semaphore(10)
    
    device_ids = [f"id-{i}" for i in range(100)]
    
    tasks = [
        fetch_device_with_limit(client, device_id, semaphore)
        for device_id in device_ids
    ]
    
    devices = await asyncio.gather(*tasks)
    print(f"Fetched {len(devices)} devices with max 10 concurrent")

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

Processing in Batches

Process items in sequential batches:

import asyncio

async def process_in_batches(
    client: InfrahubClient,
    items: list,
    batch_size: int = 10
):
    """Process items in batches."""
    results = []
    
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        
        # Process batch concurrently
        batch_tasks = [
            client.get(kind="InfraDevice", id=item_id)
            for item_id in batch
        ]
        
        batch_results = await asyncio.gather(*batch_tasks)
        results.extend(batch_results)
        
        print(f"Processed batch {i//batch_size + 1}: {len(batch)} items")
        
        # Optional delay between batches
        await asyncio.sleep(0.5)
    
    return results

# Usage
device_ids = [f"id-{i}" for i in range(100)]
devices = await process_in_batches(client, device_ids, batch_size=20)

Error Handling in Async

Handle Errors with gather

Continue processing even if some tasks fail:

import asyncio
from infrahub_sdk.exceptions import GraphQLError

async def safe_get_device(
    client: InfrahubClient,
    device_id: str
):
    """Get device with error handling."""
    try:
        return await client.get(kind="InfraDevice", id=device_id)
    except GraphQLError as e:
        print(f"Error fetching {device_id}: {e.message}")
        return None

async def fetch_all_safely():
    client = InfrahubClient()
    
    device_ids = ["id-1", "invalid-id", "id-3"]
    
    tasks = [
        safe_get_device(client, device_id)
        for device_id in device_ids
    ]
    
    devices = await asyncio.gather(*tasks)
    
    # Filter out None results
    valid_devices = [d for d in devices if d is not None]
    
    print(f"Successfully fetched {len(valid_devices)} devices")

Using return_exceptions

import asyncio

async def fetch_with_exceptions():
    client = InfrahubClient()
    
    device_ids = ["id-1", "invalid-id", "id-3"]
    
    tasks = [
        client.get(kind="InfraDevice", id=device_id)
        for device_id in device_ids
    ]
    
    # Return exceptions instead of raising
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Process results
    devices = []
    errors = []
    
    for result in results:
        if isinstance(result, Exception):
            errors.append(result)
        else:
            devices.append(result)
    
    print(f"Devices: {len(devices)}, Errors: {len(errors)}")

Async Context Managers

Using Client as Context Manager

import asyncio
from infrahub_sdk import InfrahubClient

async def use_context_manager():
    async with InfrahubClient() as client:
        devices = await client.all(kind="InfraDevice")
        print(f"Found {len(devices)} devices")
        
        # Client automatically closes on exit

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

Custom Async Context Manager

from contextlib import asynccontextmanager

@asynccontextmanager
async def device_transaction(client: InfrahubClient, branch_name: str):
    """Context manager for transactional device operations."""
    # Setup: create branch
    branch = await client.branch.create(branch_name=branch_name)
    
    try:
        yield client, branch.name
        # Commit: merge or validate
        print(f"Transaction succeeded on {branch.name}")
    
    except Exception as e:
        # Rollback: delete branch
        await client.branch.delete(branch_name=branch.name)
        print(f"Transaction rolled back: {e}")
        raise

# Usage
async with device_transaction(client, "tx-branch") as (client, branch):
    device = await client.create(
        kind="InfraDevice",
        name="new-device",
        serial_number="SN123456",
        branch=branch
    )
    await device.save()

Async Generators

Stream Results

Process items as they’re retrieved:

import asyncio
from typing import AsyncGenerator

async def stream_devices(
    client: InfrahubClient,
    batch_size: int = 10
) -> AsyncGenerator:
    """Stream devices in batches."""
    offset = 0
    
    while True:
        # Query batch using GraphQL
        query = """
        query GetDevices($offset: Int!, $limit: Int!) {
          InfraDevice(offset: $offset, limit: $limit) {
            edges {
              node {
                id
                name { value }
              }
            }
          }
        }
        """
        
        result = await client.execute_graphql(
            query=query,
            variables={"offset": offset, "limit": batch_size}
        )
        
        devices = result["InfraDevice"]["edges"]
        
        if not devices:
            break
        
        for device in devices:
            yield device["node"]
        
        offset += batch_size

# Usage
async def process_stream():
    client = InfrahubClient()
    
    count = 0
    async for device in stream_devices(client):
        print(f"Processing: {device['name']['value']}")
        count += 1
    
    print(f"Processed {count} devices")

Advanced Patterns

Task Coordination

Coordinate multiple async tasks:

import asyncio

async def coordinator():
    client = InfrahubClient()
    
    # Start multiple concurrent workflows
    task1 = asyncio.create_task(create_devices(client, 10))
    task2 = asyncio.create_task(update_devices(client))
    task3 = asyncio.create_task(query_devices(client))
    
    # Wait for all to complete
    results = await asyncio.gather(task1, task2, task3)
    
    print("All workflows completed")
    return results

async def create_devices(client, count):
    # Implementation
    pass

async def update_devices(client):
    # Implementation
    pass

async def query_devices(client):
    # Implementation
    pass

Queue-Based Processing

Use async queues for producer-consumer pattern:

import asyncio

async def producer(
    queue: asyncio.Queue,
    client: InfrahubClient
):
    """Produce device IDs."""
    devices = await client.all(kind="InfraDevice")
    
    for device in devices:
        await queue.put(device.id)
    
    # Signal completion
    await queue.put(None)

async def consumer(
    queue: asyncio.Queue,
    client: InfrahubClient,
    worker_id: int
):
    """Consume and process device IDs."""
    while True:
        device_id = await queue.get()
        
        if device_id is None:
            # Re-add sentinel for other workers
            await queue.put(None)
            break
        
        # Process device
        device = await client.get(kind="InfraDevice", id=device_id)
        print(f"Worker {worker_id}: {device.name.value}")
        
        queue.task_done()

async def queue_processing():
    client = InfrahubClient()
    queue = asyncio.Queue()
    
    # Start producer
    producer_task = asyncio.create_task(
        producer(queue, client)
    )
    
    # Start multiple consumers
    consumer_tasks = [
        asyncio.create_task(consumer(queue, client, i))
        for i in range(5)
    ]
    
    # Wait for completion
    await producer_task
    await asyncio.gather(*consumer_tasks)

Async Retry Logic

import asyncio
from functools import wraps

def async_retry(max_attempts: int = 3, delay: float = 1.0):
    """Decorator for async retry logic."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt < max_attempts - 1:
                        await asyncio.sleep(delay * (2 ** attempt))
                        continue
                    else:
                        raise
        return wrapper
    return decorator

# Usage
@async_retry(max_attempts=3, delay=1.0)
async def fetch_device(client: InfrahubClient, device_id: str):
    return await client.get(kind="InfraDevice", id=device_id)

Timeout Management

import asyncio

async def fetch_with_timeout(
    client: InfrahubClient,
    device_id: str,
    timeout: float = 5.0
):
    """Fetch device with timeout."""
    try:
        device = await asyncio.wait_for(
            client.get(kind="InfraDevice", id=device_id),
            timeout=timeout
        )
        return device
    except asyncio.TimeoutError:
        print(f"Timeout fetching device {device_id}")
        return None

Synchronous Client

When to Use Sync Client

For synchronous contexts or legacy code:

from infrahub_sdk import InfrahubClientSync

# Synchronous client (no await needed)
client = InfrahubClientSync()

# Synchronous operations
device = client.get(kind="InfrahubDevice", id="device-id")
devices = client.all(kind="InfraDevice")

device.name.value = "updated-name"
device.save()

Mixed Async/Sync

Use sync client in async code when needed:

import asyncio
from infrahub_sdk import InfrahubClient, InfrahubClientSync

async def mixed_example():
    # Async client for most operations
    async_client = InfrahubClient()
    
    devices = await async_client.all(kind="InfraDevice")
    
    # Sync client for specific operations
    sync_client = InfrahubClientSync()
    
    # Note: This blocks the event loop - use sparingly
    tag = sync_client.get(kind="BuiltinTag", id="tag-id")

Performance Optimization

Use concurrent operations

Leverage asyncio.gather() to run multiple operations concurrently.

Limit concurrency with Semaphore

Prevent overwhelming the API with too many concurrent requests.

Process in batches

Break large operations into manageable batches for better control.

Handle errors gracefully

Use return_exceptions=True or try/except to prevent one failure from stopping all operations.

Use async context managers

Ensure proper cleanup of resources with async context managers.

Next Steps

Batch Operations

Combine async with batch operations

Error Handling

Handle async errors properly

Pagination

Use async patterns for pagination

Branches

Async branch operations

Documentation Index

​Understanding Async/Await

​Why Async?

​Basic Async Pattern

​Concurrent Operations

​Query Multiple Objects Concurrently

​Create Multiple Objects Concurrently

​Update Multiple Objects Concurrently

​Controlling Concurrency

​Using Semaphore for Rate Limiting

​Processing in Batches

​Error Handling in Async

​Handle Errors with gather

​Using return_exceptions

​Async Context Managers

​Using Client as Context Manager

​Custom Async Context Manager

​Async Generators

​Stream Results

​Advanced Patterns

​Task Coordination

​Queue-Based Processing

​Async Retry Logic

​Timeout Management

​Synchronous Client

​When to Use Sync Client

​Mixed Async/Sync

​Performance Optimization

​Next Steps