Quick Start

Get started with the Infrahub Python SDK in minutes

This guide walks you through creating your first script using the Infrahub Python SDK.

Prerequisites

• Python 3.12+ installed
• Infrahub SDK installed (see Installation)
• Access to an Infrahub instance
• API token (optional, for authentication)

Basic Example

Create a file named quickstart.py:

import asyncio
from infrahub_sdk import InfrahubClient

async def main():
    # Initialize the client
    client = InfrahubClient()
    
    # Query all devices
    devices = await client.all(kind="InfraDevice")
    
    print(f"Found {len(devices)} devices:")
    for device in devices:
        print(f"  - {device.name.value}")

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

Run the script:

python quickstart.py

Configuration

Using Environment Variables

Set environment variables before running:

export INFRAHUB_ADDRESS="http://localhost:8000"
export INFRAHUB_API_TOKEN="your-token-here"
python quickstart.py

Using Configuration Object

Pass configuration directly to the client:

from infrahub_sdk import Config, InfrahubClient

client = InfrahubClient(
    config=Config(
        address="http://localhost:8000",
        api_token="your-token-here"
    )
)

Using Configuration File

Create infrahub.toml:

[sdk]
address = "http://localhost:8000"
api_token = "your-api-token"

The client automatically loads this file:

client = InfrahubClient()  # Loads infrahub.toml

Creating Objects

Create a new device:

import asyncio
from infrahub_sdk import InfrahubClient

async def create_device():
    client = InfrahubClient()
    
    # Create a device
    device = await client.create(
        kind="InfraDevice",
        name="router-01",
        description="Edge router in NYC",
        status="active"
    )
    
    # Save to Infrahub
    await device.save()
    
    print(f"Created device: {device.name.value} (ID: {device.id})")

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

Querying Data

Get a Single Object

async def get_device():
    client = InfrahubClient()
    
    # Get by ID
    device = await client.get(kind="InfraDevice", id="abc-123")
    
    # Get by filter
    device = await client.get(
        kind="InfraDevice",
        name__value="router-01"
    )
    
    if device:
        print(f"Found: {device.name.value}")
    else:
        print("Device not found")

Get All Objects

async def list_devices():
    client = InfrahubClient()
    
    # Get all devices
    devices = await client.all(kind="InfraDevice")
    
    for device in devices:
        print(f"{device.name.value} - {device.status.value}")

Filter Results

async def filter_devices():
    client = InfrahubClient()
    
    # Filter by status
    active_devices = await client.filters(
        kind="InfraDevice",
        status__value="active"
    )
    
    print(f"Found {len(active_devices)} active devices")

Updating Objects

Update an existing object:

async def update_device():
    client = InfrahubClient()
    
    # Get the device
    device = await client.get(
        kind="InfraDevice",
        name__value="router-01"
    )
    
    if device:
        # Update attributes
        device.description.value = "Updated description"
        device.status.value = "maintenance"
        
        # Save changes
        await device.save()
        print(f"Updated {device.name.value}")

Working with Relationships

Create objects with relationships:

async def create_with_relationships():
    client = InfrahubClient()
    
    # Get related objects
    site = await client.get(kind="InfraSite", name__value="NYC")
    role = await client.get(kind="BuiltinRole", name__value="edge")
    
    # Create device with relationships
    device = await client.create(
        kind="InfraDevice",
        name="router-02",
        site=site,
        role=role
    )
    
    await device.save()
    print(f"Created {device.name.value} at {site.name.value}")

Using Branches

Work with different branches:

async def work_with_branches():
    client = InfrahubClient()
    
    # List all branches
    branches = await client.branch.all()
    for branch in branches:
        print(f"Branch: {branch.name}")
    
    # Work in a specific branch
    device = await client.create(
        branch="feature-branch",
        kind="InfraDevice",
        name="test-device"
    )
    await device.save()

Error Handling

Handle errors gracefully:

from infrahub_sdk.exceptions import GraphQLError, Error

async def safe_create():
    client = InfrahubClient()
    
    try:
        device = await client.create(
            kind="InfraDevice",
            name="router-03"
        )
        await device.save()
    except GraphQLError as e:
        print(f"GraphQL error: {e.message}")
    except Error as e:
        print(f"SDK error: {e.message}")

Using the CLI Tool

The SDK includes infrahubctl for running scripts:

# Run a Python script with Infrahub context
infrahubctl run your_script.py

# List available branches
infrahubctl branch list

# Load a schema
infrahubctl schema load schema.yml

Complete Example

Here’s a complete example that creates a site and devices:

import asyncio
from infrahub_sdk import InfrahubClient
from infrahub_sdk.exceptions import GraphQLError

async def setup_infrastructure():
    client = InfrahubClient()
    
    try:
        # Create a site
        site = await client.create(
            kind="InfraSite",
            name="NYC",
            description="New York City datacenter"
        )
        await site.save()
        print(f"Created site: {site.name.value}")
        
        # Create multiple devices
        batch = await client.create_batch()
        
        for i in range(1, 4):
            device = await client.create(
                kind="InfraDevice",
                name=f"router-{i:02d}",
                site=site,
                status="active"
            )
            batch.add(task=device.save, node=device)
        
        # Execute batch
        async for node, result in batch.execute():
            print(f"Created device: {node.name.value}")
        
        print("Infrastructure setup complete!")
        
    except GraphQLError as e:
        print(f"Error: {e.message}")

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

Next Steps

Client Configuration

Learn more about configuring the InfrahubClient

Queries

Advanced querying techniques

Mutations

Creating and updating data

Batch Operations

Efficient bulk operations