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

Build docs developers (and LLMs) love

Get started for freeTalk to us