Branches - Infrahub

Infrahub provides Git-like branching for infrastructure data, allowing you to make changes in isolation before merging them to the main branch.

Understanding Branches

Branches in Infrahub work similarly to Git:

Main branch: The default, production branch
Feature branches: Isolated workspaces for changes
Merging: Combine branch changes back to main

All objects and changes are branch-aware. You can query and modify data on specific branches.

Creating Branches

Create a New Branch

from infrahub_sdk import InfrahubClient

client = InfrahubClient()

# Create a feature branch
branch = await client.branch.create(branch_name="feature-update")

print(f"Created branch: {branch.name}")
print(f"Branch ID: {branch.id}")
print(f"Created at: {branch.created_at}")

Create Branch with Description

branch = await client.branch.create(
    branch_name="feature-new-datacenter",
    description="Add new datacenter infrastructure",
    sync_with_git=True
)

print(f"Branch: {branch.name}")
print(f"Description: {branch.description}")

Create from Specific Base

# Create branch from another branch
branch = await client.branch.create(
    branch_name="feature-sub-branch",
    branch_from="feature-parent-branch"
)

Working on Branches

Create Objects on a Branch

# Create a branch
branch = await client.branch.create(branch_name="add-devices")

# Create objects on that branch
device = await client.create(
    kind="InfraDevice",
    name="router-new",
    serial_number="SN123456",
    branch=branch.name  # Specify the branch
)
await device.save()

print(f"Created device on branch: {branch.name}")

Query Objects on a Branch

# Query on a specific branch
devices_on_branch = await client.all(
    kind="InfraDevice",
    branch="feature-update"
)

print(f"Devices on feature-update: {len(devices_on_branch)}")

Update Objects on a Branch

# Create branch
branch = await client.branch.create(branch_name="update-config")

# Get object on that branch
device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch=branch.name
)

# Update on the branch
device.name.value = "router-updated"
await device.save()

print(f"Updated device on branch: {branch.name}")

Delete Objects on a Branch

branch = await client.branch.create(branch_name="cleanup")

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch=branch.name
)

await device.delete()
print(f"Deleted device on branch: {branch.name}")

Querying Branches

List All Branches

branches = await client.branch.all()

print(f"Total branches: {len(branches)}")
for branch in branches:
    print(f"  - {branch.name}")

Get Branch Details

branch = await client.branch.get(branch_name="feature-update")

print(f"Name: {branch.name}")
print(f"Description: {branch.description}")
print(f"Created at: {branch.created_at}")
print(f"Is default: {branch.is_default}")

Check Branch Status

branch = await client.branch.get(branch_name="feature-update")

if branch.has_schema_changes:
    print("Branch has schema changes")

if branch.is_isolated:
    print("Branch is isolated")

Comparing Branches

Compare Object Across Branches

# Get object on main branch
main_device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch="main"
)

# Get same object on feature branch
feature_device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch="feature-update"
)

# Compare values
if main_device.name.value != feature_device.name.value:
    print(f"Name changed:")
    print(f"  Main: {main_device.name.value}")
    print(f"  Feature: {feature_device.name.value}")

Get Branch Diff

# Get differences between branches
diff = await client.branch.diff(
    branch_name="feature-update",
    target_branch="main"
)

print(f"Added: {len(diff.added)}")
print(f"Modified: {len(diff.modified)}")
print(f"Deleted: {len(diff.deleted)}")

Count Changes on Branch

branch = await client.branch.get(branch_name="feature-update")

# Get all objects on branch
branch_devices = await client.all(
    kind="InfraDevice",
    branch=branch.name
)

# Get objects on main
main_devices = await client.all(
    kind="InfraDevice",
    branch="main"
)

print(f"Main: {len(main_devices)} devices")
print(f"Branch: {len(branch_devices)} devices")
print(f"Difference: {len(branch_devices) - len(main_devices)}")

Branch Workflows

Feature Development Workflow

import asyncio
from infrahub_sdk import InfrahubClient

async def feature_workflow():
    client = InfrahubClient()
    
    # 1. Create feature branch
    branch = await client.branch.create(
        branch_name="feature-new-location",
        description="Add new datacenter location"
    )
    print(f"Created branch: {branch.name}")
    
    # 2. Make changes on the branch
    location = await client.create(
        kind="InfraLocation",
        name="Datacenter-NYC",
        address="New York",
        branch=branch.name
    )
    await location.save()
    print(f"Created location on branch")
    
    # 3. Create related objects
    for i in range(3):
        device = await client.create(
            kind="InfraDevice",
            name=f"nyc-router-{i:02d}",
            serial_number=f"SN{i:06d}",
            location=location,
            branch=branch.name
        )
        await device.save()
    
    print(f"Created 3 devices on branch")
    
    # 4. Verify changes
    branch_devices = await client.all(
        kind="InfraDevice",
        branch=branch.name
    )
    print(f"Total devices on branch: {len(branch_devices)}")
    
    # 5. Ready to merge (in UI or via API)
    print(f"Branch {branch.name} ready for review")

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

Testing Changes Workflow

async def test_changes_workflow():
    client = InfrahubClient()
    
    # Create test branch
    test_branch = await client.branch.create(
        branch_name="test-configuration-change"
    )
    
    # Make changes
    devices = await client.all(
        kind="InfraDevice",
        branch=test_branch.name
    )
    
    for device in devices:
        device_full = await client.get(
            kind="InfraDevice",
            id=device.id,
            branch=test_branch.name
        )
        device_full.is_managed.value = True
        await device_full.save()
    
    print(f"Updated {len(devices)} devices on test branch")
    
    # Run tests/validations
    validation_passed = True  # Your validation logic
    
    if validation_passed:
        print("Tests passed - ready to merge")
    else:
        print("Tests failed - delete branch")
        await client.branch.delete(branch_name=test_branch.name)

Bulk Update Workflow

async def bulk_update_workflow():
    client = InfrahubClient()
    
    # Create update branch
    branch = await client.branch.create(
        branch_name="bulk-update-locations"
    )
    
    # Get all devices on branch
    devices = await client.all(
        kind="InfraDevice",
        branch=branch.name
    )
    
    # Get new location
    new_location = await client.get(
        kind="InfraLocation",
        id="new-location-id",
        branch=branch.name
    )
    
    # Update all devices
    for device in devices:
        device_full = await client.get(
            kind="InfraDevice",
            id=device.id,
            branch=branch.name
        )
        device_full.location = new_location
        await device_full.save()
    
    print(f"Updated {len(devices)} devices on branch {branch.name}")

Branch Management

Delete a Branch

# Delete a branch
await client.branch.delete(branch_name="feature-old")

print("Branch deleted")

Rebase a Branch

# Rebase branch onto main
await client.branch.rebase(
    branch_name="feature-update",
    target_branch="main"
)

print("Branch rebased")

Validate Branch

# Check if branch can be merged
validation = await client.branch.validate(
    branch_name="feature-update"
)

if validation.is_valid:
    print("Branch is valid and can be merged")
else:
    print(f"Validation errors: {validation.errors}")

Schema Changes on Branches

Load Schema on Branch

branch = await client.branch.create(
    branch_name="schema-update"
)

# Load new schema on branch
schema = {
    "version": "1.0",
    "nodes": [
        {
            "name": "NewDevice",
            "namespace": "Infra",
            "attributes": [
                {"name": "name", "kind": "Text"},
                {"name": "model", "kind": "Text"}
            ]
        }
    ]
}

response = await client.schema.load(
    schemas=[schema],
    branch=branch.name,
    wait_until_converged=True
)

if response.schema_updated:
    print("Schema updated on branch")

Query Schema on Branch

# Get schema on specific branch
device_schema = await client.schema.get(
    kind="InfraDevice",
    branch="feature-schema-update"
)

print(f"Schema on branch: {device_schema.name}")
print(f"Attributes: {[attr.name for attr in device_schema.attributes]}")

Advanced Branch Patterns

Multi-Branch Coordination

async def multi_branch_workflow():
    client = InfrahubClient()
    
    # Create multiple feature branches
    branches = []
    for i in range(3):
        branch = await client.branch.create(
            branch_name=f"feature-team-{i+1}"
        )
        branches.append(branch)
    
    # Each team works on their branch
    for idx, branch in enumerate(branches):
        device = await client.create(
            kind="InfraDevice",
            name=f"team-{idx+1}-device",
            serial_number=f"SN{idx:06d}",
            branch=branch.name
        )
        await device.save()
        print(f"Team {idx+1} created device on {branch.name}")
    
    # Merge branches sequentially
    for branch in branches:
        print(f"Ready to merge: {branch.name}")
        # Merge would happen via UI or API

Branch-Based Testing

async def branch_testing():
    client = InfrahubClient()
    
    # Create test branch
    test_branch = await client.branch.create(
        branch_name="test-deployment"
    )
    
    try:
        # Make risky changes
        devices = await client.all(
            kind="InfraDevice",
            branch=test_branch.name
        )
        
        for device in devices:
            device_full = await client.get(
                kind="InfraDevice",
                id=device.id,
                branch=test_branch.name
            )
            device_full.firmware_version.value = "2.0.0"
            await device_full.save()
        
        # Run validations
        all_valid = True  # Your validation logic
        
        if all_valid:
            print("Tests passed - merge branch")
            # Merge logic
        else:
            raise Exception("Validation failed")
    
    except Exception as e:
        print(f"Test failed: {e}")
        # Delete test branch
        await client.branch.delete(branch_name=test_branch.name)
        print("Test branch deleted")

Branch Isolation Testing

async def test_isolation():
    client = InfrahubClient()
    
    # Create isolated branch
    branch = await client.branch.create(
        branch_name="isolated-test"
    )
    
    # Create object on branch
    device = await client.create(
        kind="InfraDevice",
        name="test-device",
        serial_number="SN000000",
        branch=branch.name
    )
    await device.save()
    
    # Verify it exists on branch
    branch_devices = await client.all(
        kind="InfraDevice",
        branch=branch.name
    )
    print(f"Devices on branch: {len(branch_devices)}")
    
    # Verify it doesn't exist on main
    main_devices = await client.all(
        kind="InfraDevice",
        branch="main"
    )
    print(f"Devices on main: {len(main_devices)}")
    
    # Cleanup
    await client.branch.delete(branch_name=branch.name)

Best Practices

Use descriptive branch names

Name branches to clearly indicate their purpose:

feature-add-datacenter-nyc
fix-device-configuration
update-schema-v2

Keep branches short-lived

Merge or delete branches frequently to avoid drift from main.

Test on branches

Use branches as safe environments for testing changes before merging to production.

Clean up merged branches

Delete branches after merging to keep your branch list manageable.

Next Steps

Batch Operations

Perform efficient bulk operations

Error Handling

Handle branch-related errors

Async Operations

Use async patterns with branches

Pagination

Handle large datasets on branches

Getting Started

Usage Guide

Advanced

​Understanding Branches

​Creating Branches

​Create a New Branch

​Create Branch with Description

​Create from Specific Base

​Working on Branches

​Create Objects on a Branch

​Query Objects on a Branch

​Update Objects on a Branch

​Delete Objects on a Branch

​Querying Branches

​List All Branches

​Get Branch Details

​Check Branch Status

​Comparing Branches

​Compare Object Across Branches

​Get Branch Diff

​Count Changes on Branch

​Branch Workflows

​Feature Development Workflow

​Testing Changes Workflow

​Bulk Update Workflow

​Branch Management

​Delete a Branch

​Rebase a Branch

​Validate Branch

​Schema Changes on Branches

​Load Schema on Branch

​Query Schema on Branch

​Advanced Branch Patterns

​Multi-Branch Coordination

​Branch-Based Testing

​Branch Isolation Testing

​Best Practices

​Next Steps

Batch Operations

Error Handling

Async Operations

Pagination

Build docs developers (and LLMs) love

Understanding Branches

Creating Branches

Create a New Branch

Create Branch with Description

Create from Specific Base

Working on Branches

Create Objects on a Branch

Query Objects on a Branch

Update Objects on a Branch

Delete Objects on a Branch

Querying Branches

List All Branches

Get Branch Details

Check Branch Status

Comparing Branches

Compare Object Across Branches

Get Branch Diff

Count Changes on Branch

Branch Workflows

Feature Development Workflow

Testing Changes Workflow

Bulk Update Workflow

Branch Management

Delete a Branch

Rebase a Branch

Validate Branch

Schema Changes on Branches

Load Schema on Branch

Query Schema on Branch

Advanced Branch Patterns

Multi-Branch Coordination

Branch-Based Testing

Branch Isolation Testing

Best Practices

Next Steps