Skip to main content
This guide walks you through creating your first application using the Infrahub Python SDK.

Prerequisites

1

Install Python 3.12+

Ensure you have Python 3.12 or higher installed:
python --version
2

Install the SDK

Install the Infrahub SDK:
pip install infrahub-sdk
3

Verify Infrahub is Running

Ensure your Infrahub instance is accessible:
curl http://localhost:8000/api/

Your First Script

Create a file called quickstart.py:
quickstart.py
import asyncio
from infrahub_sdk import InfrahubClient

async def main():
    # Initialize the client
    client = InfrahubClient()
    
    # Create a tag
    tag = await client.create(
        kind="BuiltinTag",
        name="production",
        description="Production environment"
    )
    await tag.save()
    print(f"Created tag: {tag.name.value} (ID: {tag.id})")
    
    # Query the tag back
    retrieved_tag = await client.get(
        kind="BuiltinTag",
        id=tag.id
    )
    print(f"Retrieved tag: {retrieved_tag.name.value}")
    
    # Update the tag
    retrieved_tag.description.value = "Production environment - updated"
    await retrieved_tag.save()
    print(f"Updated description: {retrieved_tag.description.value}")
    
    # Query all tags
    all_tags = await client.all(kind="BuiltinTag")
    print(f"Total tags: {len(all_tags)}")
    for t in all_tags:
        print(f"  - {t.name.value}")

if __name__ == "__main__":
    asyncio.run(main())
Run the script:
python quickstart.py
The SDK is async-first, so all I/O operations use async/await. Always run your code with asyncio.run().

Understanding the Code

1. Import and Initialize

from infrahub_sdk import InfrahubClient

client = InfrahubClient()
The InfrahubClient is your main interface to Infrahub. By default, it connects to http://localhost:8000.

2. Create an Object

tag = await client.create(
    kind="BuiltinTag",
    name="production",
    description="Production environment"
)
await tag.save()
  • create() instantiates a new object in memory
  • save() persists it to Infrahub via the API
  • The kind parameter specifies the schema type

3. Query an Object

retrieved_tag = await client.get(
    kind="BuiltinTag",
    id=tag.id
)
Fetch an object by its kind and ID.

4. Update an Object

retrieved_tag.description.value = "Production environment - updated"
await retrieved_tag.save()
Modify attributes and call save() to persist changes.

5. Query All Objects

all_tags = await client.all(kind="BuiltinTag")
Retrieve all objects of a specific kind.

Working with Relationships

Create objects with relationships:
import asyncio
from infrahub_sdk import InfrahubClient

async def create_with_relationships():
    client = InfrahubClient()
    
    # Load a schema first (example)
    schema = {
        "version": "1.0",
        "nodes": [
            {
                "name": "Device",
                "namespace": "Infra",
                "attributes": [
                    {"name": "name", "kind": "Text"},
                    {"name": "serial_number", "kind": "Text"}
                ],
                "relationships": [
                    {
                        "name": "location",
                        "peer": "InfraLocation",
                        "kind": "Attribute",
                        "cardinality": "one"
                    }
                ]
            },
            {
                "name": "Location",
                "namespace": "Infra",
                "attributes": [
                    {"name": "name", "kind": "Text"},
                    {"name": "address", "kind": "Text", "optional": True}
                ]
            }
        ]
    }
    
    await client.schema.load(schemas=[schema])
    
    # Create a location
    location = await client.create(
        kind="InfraLocation",
        name="Datacenter-1",
        address="123 Main St"
    )
    await location.save()
    
    # Create a device with a relationship to the location
    device = await client.create(
        kind="InfraDevice",
        name="router-01",
        serial_number="SN123456",
        location=location  # Pass the related object
    )
    await device.save()
    
    print(f"Created device '{device.name.value}' at location '{location.name.value}'")
    
    # Query back with relationships
    device_with_location = await client.get(
        kind="InfraDevice",
        id=device.id,
        include=["location"]  # Include related data
    )
    
    print(f"Device location: {device_with_location.location.name.value}")

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

Working with Branches

Infrahub supports Git-like branching:
import asyncio
from infrahub_sdk import InfrahubClient

async def branch_workflow():
    client = InfrahubClient()
    
    # Create a new branch
    branch = await client.branch.create(branch_name="feature-update")
    print(f"Created branch: {branch.name}")
    
    # Work on the branch
    tag = await client.create(
        kind="BuiltinTag",
        name="staging",
        branch=branch.name  # Specify the branch
    )
    await tag.save()
    
    print(f"Created tag on branch '{branch.name}'")
    
    # Query on the branch
    tags_on_branch = await client.all(
        kind="BuiltinTag",
        branch=branch.name
    )
    print(f"Tags on branch: {len(tags_on_branch)}")

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

Error Handling

Handle errors gracefully:
import asyncio
from infrahub_sdk import InfrahubClient
from infrahub_sdk.exceptions import GraphQLError, NodeNotFoundError

async def error_handling_example():
    client = InfrahubClient()
    
    try:
        # Try to get a non-existent object
        tag = await client.get(kind="BuiltinTag", id="invalid-id")
    except NodeNotFoundError:
        print("Tag not found")
    except GraphQLError as e:
        print(f"GraphQL error: {e.message}")
    except Exception as e:
        print(f"Unexpected error: {e}")

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

Configuration

Configure the client for different environments:
from infrahub_sdk import Config, InfrahubClient

# Production configuration
config = Config(
    address="https://infrahub.example.com",
    api_token="your-api-token",
    timeout=60
)

client = InfrahubClient(config=config)
Or use environment variables:
export INFRAHUB_ADDRESS="https://infrahub.example.com"
export INFRAHUB_API_TOKEN="your-api-token"
from infrahub_sdk import InfrahubClient

# Automatically uses environment variables
client = InfrahubClient()

Common Patterns

Bulk Creation

devices = []
for i in range(10):
    device = await client.create(
        kind="InfraDevice",
        name=f"device-{i:02d}",
        serial_number=f"SN{i:06d}"
    )
    await device.save()
    devices.append(device)

print(f"Created {len(devices)} devices")

Filtering and Searching

# Get all devices with specific properties
all_devices = await client.all(kind="InfraDevice")

# Filter in Python
production_devices = [
    d for d in all_devices 
    if "prod" in d.name.value.lower()
]

Accessing Nested Relationships

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["location", "tags"]
)

print(f"Device: {device.name.value}")
print(f"Location: {device.location.name.value}")
print(f"Tags: {[tag.name.value for tag in device.tags]}")

Next Steps

Client Setup

Learn advanced client configuration

Querying Data

Master data querying techniques

Creating Objects

Deep dive into object creation

Error Handling

Handle errors and exceptions properly