InfrahubClient - Infrahub

The InfrahubClient is the main entry point for interacting with Infrahub’s API. It handles authentication, GraphQL queries, and provides methods for CRUD operations.

Basic Usage

from infrahub_sdk import InfrahubClient

client = InfrahubClient()

Configuration

Using Config Object

Configure the client explicitly:

from infrahub_sdk import Config, InfrahubClient

client = InfrahubClient(
    config=Config(
        address="http://localhost:8000",
        api_token="your-api-token",
        timeout=30,
        retry_on_failure=True,
        max_concurrent_execution=5
    )
)

Configuration Options

Parameter	Type	Default	Description
`address`	str	`http://localhost:8000`	Infrahub API URL
`api_token`	str	None	API authentication token
`username`	str	None	Username (alternative to token)
`password`	str	None	Password (alternative to token)
`timeout`	int	60	Request timeout in seconds
`retry_on_failure`	bool	False	Retry failed requests
`max_concurrent_execution`	int	10	Max parallel requests
`insert_tracker`	bool	True	Track request metadata

Environment Variables

The client reads these environment variables:

INFRAHUB_ADDRESS="http://localhost:8000"
INFRAHUB_API_TOKEN="your-token"
INFRAHUB_USERNAME="admin"
INFRAHUB_PASSWORD="admin-password"
INFRAHUB_TIMEOUT="60"

Configuration File

Create infrahub.toml in your project root:

[sdk]
address = "http://localhost:8000"
api_token = "your-api-token"
timeout = 60
retry_on_failure = true
max_concurrent_execution = 10

The client automatically loads this file if it exists.

Authentication

API Token (Recommended)

Use an API token for authentication:

from infrahub_sdk import Config, InfrahubClient

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

Generate a token in the Infrahub UI under Account Settings.

Username and Password

Alternatively, use username and password:

client = InfrahubClient(
    config=Config(
        address="http://localhost:8000",
        username="admin",
        password="admin-password"
    )
)

Username/password authentication is less secure than API tokens. Use tokens in production.

Client Methods

Query Operations

get()

Retrieve a single object:

# 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"
)

# Get with specific branch
device = await client.get(
    branch="feature-branch",
    kind="InfraDevice",
    id="abc-123"
)

all()

Retrieve all objects of a kind:

# Get all devices
devices = await client.all(kind="InfraDevice")

# Get all devices in a branch
devices = await client.all(
    branch="feature-branch",
    kind="InfraDevice"
)

filters()

Retrieve objects with filters:

# Filter by attribute
devices = await client.filters(
    kind="InfraDevice",
    status__value="active"
)

# Multiple filters
devices = await client.filters(
    kind="InfraDevice",
    status__value="active",
    role__name__value="edge"
)

Mutation Operations

create()

Create a new object:

device = await client.create(
    kind="InfraDevice",
    name="router-01",
    description="Edge router",
    status="active"
)

# Note: Must call save() to persist
await device.save()

delete()

Delete an object:

# Get the object first
device = await client.get(kind="InfraDevice", id="abc-123")

# Delete it
await client.delete(node=device)

Schema Operations

schema.load()

Load a schema into Infrahub:

schema = {
    "version": "1.0",
    "nodes": [
        {
            "name": "Device",
            "namespace": "Infra",
            "attributes": [
                {"name": "name", "kind": "Text"},
                {"name": "status", "kind": "Text"}
            ]
        }
    ]
}

response = await client.schema.load(schemas=[schema])
if response.errors:
    print(f"Errors: {response.errors}")

schema.get()

Retrieve schema for a kind:

schema = await client.schema.get(kind="InfraDevice")
print(f"Attributes: {schema.attributes}")

Branch Operations

branch.all()

List all branches:

branches = await client.branch.all()
for branch in branches:
    print(f"{branch.name} - {branch.description}")

branch.create()

Create a new branch:

branch = await client.branch.create(
    branch_name="feature-new-devices",
    description="Adding new devices",
    sync_with_git=False
)

branch.merge()

Merge a branch:

await client.branch.merge(
    branch_name="feature-new-devices"
)

branch.delete()

Delete a branch:

await client.branch.delete(
    branch_name="feature-new-devices"
)

Batch Operations

create_batch()

Create a batch for multiple operations:

batch = await client.create_batch()

for i in range(10):
    device = await client.create(
        kind="InfraDevice",
        name=f"router-{i:02d}"
    )
    batch.add(task=device.save, node=device)

async for node, result in batch.execute():
    print(f"Created: {node.name.value}")

See Batch Operations for more details.

Advanced Configuration

Custom Headers

Add custom HTTP headers:

from infrahub_sdk import Config, InfrahubClient

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

client = InfrahubClient(config=config)
client.headers["X-Custom-Header"] = "value"

Logging

Enable SDK logging:

import logging
from infrahub_sdk import Config, InfrahubClient

logger = logging.getLogger("infrahub_sdk")
logger.setLevel(logging.DEBUG)

client = InfrahubClient(
    config=Config(
        address="http://localhost:8000",
        log=logger
    )
)

Retry Configuration

Configure retry behavior:

from infrahub_sdk import Config, InfrahubClient

client = InfrahubClient(
    config=Config(
        address="http://localhost:8000",
        retry_on_failure=True,
        retry_delay=5,  # seconds between retries
        max_retries=3
    )
)

Timeout Configuration

Set different timeouts:

client = InfrahubClient(
    config=Config(
        address="http://localhost:8000",
        timeout=120,  # Overall timeout
        connect_timeout=10,  # Connection timeout
    )
)

Context Manager

Use the client as a context manager for automatic cleanup:

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

Working with Multiple Clients

Create multiple clients for different Infrahub instances:

prod_client = InfrahubClient(
    config=Config(address="https://prod.infrahub.example.com")
)

dev_client = InfrahubClient(
    config=Config(address="https://dev.infrahub.example.com")
)

# Copy data from prod to dev
prod_devices = await prod_client.all(kind="InfraDevice")
for device in prod_devices:
    new_device = await dev_client.create(
        kind="InfraDevice",
        name=device.name.value
    )
    await new_device.save()

Error Handling

Handle common errors:

from infrahub_sdk.exceptions import (
    GraphQLError,
    Error,
    AuthenticationError,
    ServerNotReachableError
)

try:
    client = InfrahubClient(
        config=Config(
            address="http://localhost:8000",
            api_token="invalid-token"
        )
    )
    await client.branch.all()
except AuthenticationError:
    print("Invalid credentials")
except ServerNotReachableError:
    print("Cannot reach Infrahub server")
except Error as e:
    print(f"SDK error: {e.message}")

Performance Tips

Connection Pooling

The client uses connection pooling by default. Configure pool size:

client = InfrahubClient(
    config=Config(
        address="http://localhost:8000",
        max_concurrent_execution=20  # Increase for more parallelism
    )
)

Batch Operations

Use batch operations for bulk creates/updates:

# Slow - individual saves
for i in range(100):
    device = await client.create(kind="InfraDevice", name=f"dev-{i}")
    await device.save()  # 100 API calls

# Fast - batch operation
batch = await client.create_batch()
for i in range(100):
    device = await client.create(kind="InfraDevice", name=f"dev-{i}")
    batch.add(task=device.save, node=device)
async for node, result in batch.execute():  # Optimized API calls
    pass

Field Selection

Limit fields returned to reduce payload size:

from infrahub_sdk.graphql import Query

query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None}
                    # Only fetch id and name
                }
            }
        }
    }
)

result = await client.execute_graphql(query=query.render_query())

Next Steps

Queries

Learn advanced querying techniques

Mutations

Create and update data

Batch Operations

Efficient bulk operations

Documentation Index

​Basic Usage

​Configuration

​Using Config Object

​Configuration Options

​Environment Variables

​Configuration File

​Authentication

​API Token (Recommended)

​Username and Password

​Client Methods

​Query Operations

​get()

​all()

​filters()

​Mutation Operations

​create()

​delete()

​Schema Operations

​schema.load()

​schema.get()

​Branch Operations

​branch.all()

​branch.create()

​branch.merge()

​branch.delete()

​Batch Operations

​create_batch()

​Advanced Configuration

​Custom Headers

​Logging

​Retry Configuration

​Timeout Configuration

​Context Manager

​Working with Multiple Clients

​Error Handling

​Performance Tips

​Connection Pooling

​Batch Operations

​Field Selection

​Next Steps

Queries

Mutations

Batch Operations

Basic Usage

Configuration

Using Config Object

Configuration Options

Environment Variables

Configuration File

Authentication

API Token (Recommended)

Username and Password

Client Methods

Query Operations

get()

all()

filters()

Mutation Operations

create()

delete()

Schema Operations

schema.load()

schema.get()

Branch Operations

branch.all()

branch.create()

branch.merge()

branch.delete()

Batch Operations

create_batch()

Advanced Configuration

Custom Headers

Logging

Retry Configuration

Timeout Configuration

Context Manager

Working with Multiple Clients

Error Handling

Performance Tips

Connection Pooling

Batch Operations

Field Selection

Next Steps