Mutations - Infrahub

Mutations in the Infrahub SDK allow you to create, update, and delete objects in your infrastructure data model.

Creating Objects

Basic Create

Create a new object:

from infrahub_sdk import InfrahubClient

client = InfrahubClient()

# Create the object
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 with ID: {device.id}")

Remember to call save() after creating an object. The create() method only initializes the object locally.

Create with Relationships

Create objects with related entities:

# Get related objects first
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,
    status="active"
)

await device.save()

Create in a Branch

Create objects in a specific branch:

device = await client.create(
    branch="feature-new-devices",
    kind="InfraDevice",
    name="router-03",
    status="active"
)

await device.save()

Create with Protected Values

Mark attribute values as protected:

device = await client.create(
    kind="InfraDevice",
    name={
        "value": "router-04",
        "is_protected": True  # Protect from changes
    },
    status="active"
)

await device.save()

Create with Source Tracking

Track the source of attribute values:

# Get source account
account = await client.get(kind="CoreAccount", name="admin")

device = await client.create(
    kind="InfraDevice",
    name={
        "value": "router-05",
        "source": account.id  # Track who set this value
    },
    status="active"
)

await device.save()

Updating Objects

Basic Update

Update an existing object:

# Get the object
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}")

Update Multiple Attributes

Update several attributes at once:

device = await client.get(kind="InfraDevice", id="abc-123")

if device:
    device.name.value = "router-01-updated"
    device.description.value = "Updated router"
    device.status.value = "active"
    
    await device.save()

Update Relationships

Change relationships to other objects:

device = await client.get(kind="InfraDevice", id="abc-123")

# Get new related objects
new_site = await client.get(kind="InfraSite", name__value="LAX")
new_role = await client.get(kind="BuiltinRole", name__value="core")

if device:
    # Update relationships
    device.site.peer = new_site
    device.role.peer = new_role
    
    await device.save()

Update in a Branch

Make updates in a specific branch:

device = await client.get(
    branch="feature-updates",
    kind="InfraDevice",
    id="abc-123"
)

if device:
    device.status.value = "maintenance"
    await device.save()

Protect Attributes

Protect attribute values from changes:

device = await client.get(kind="InfraDevice", id="abc-123")

if device:
    device.name.value = "router-01-prod"
    device.name.is_protected = True  # Protect from future changes
    
    await device.save()

Deleting Objects

Basic Delete

Delete an object:

# Get the object
device = await client.get(
    kind="InfraDevice",
    name__value="router-test"
)

if device:
    # Delete it
    await client.delete(node=device)
    print(f"Deleted {device.name.value}")

Delete by ID

Delete using object ID:

device = await client.get(kind="InfraDevice", id="abc-123")
if device:
    await client.delete(node=device)

Delete in a Branch

Delete objects in a specific branch:

device = await client.get(
    branch="feature-cleanup",
    kind="InfraDevice",
    id="abc-123"
)

if device:
    await client.delete(node=device)

Working with Many Relationships

Add to Many Relationship

Add items to a many relationship:

device = await client.get(kind="InfraDevice", id="abc-123")
tag1 = await client.get(kind="BuiltinTag", name__value="production")
tag2 = await client.get(kind="BuiltinTag", name__value="critical")

if device:
    # Add tags (many relationship)
    device.tags.peers = [tag1, tag2]
    await device.save()

Update Many Relationship

Replace all items in a many relationship:

device = await client.get(kind="InfraDevice", id="abc-123")

# Get new tags
new_tags = await client.filters(
    kind="BuiltinTag",
    name__value__in=["production", "network", "edge"]
)

if device:
    # Replace all tags
    device.tags.peers = new_tags
    await device.save()

Remove from Many Relationship

Remove specific items:

device = await client.get(kind="InfraDevice", id="abc-123")

if device:
    # Get current tags
    current_tags = await device.tags.fetch()
    
    # Filter out tags to remove
    new_tags = [tag for tag in current_tags if tag.name.value != "old-tag"]
    
    # Update
    device.tags.peers = new_tags
    await device.save()

GraphQL Mutations

Custom Mutations

Build custom GraphQL mutations:

from infrahub_sdk.graphql import Mutation

# Define a custom mutation
mutation = Mutation(
    mutation={
        "InfraDeviceCreate": {
            "__args": {
                "data": {
                    "name": {"value": "router-06"},
                    "status": {"value": "active"}
                }
            },
            "object": {
                "id": None,
                "name": {"value": None}
            }
        }
    }
)

# Execute the mutation
result = await client.execute_graphql(query=mutation.render_query())

Upsert Pattern

Create or update based on existence:

async def upsert_device(name: str, **attributes):
    client = InfrahubClient()
    
    # Try to get existing
    device = await client.get(
        kind="InfraDevice",
        name__value=name
    )
    
    if device:
        # Update existing
        for attr, value in attributes.items():
            setattr(getattr(device, attr), 'value', value)
    else:
        # Create new
        device = await client.create(
            kind="InfraDevice",
            name=name,
            **attributes
        )
    
    await device.save()
    return device

# Usage
device = await upsert_device(
    name="router-01",
    status="active",
    description="Updated or created"
)

Bulk Operations

Create Multiple Objects

Create many objects efficiently:

# Create a batch
batch = await client.create_batch()

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

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

See Batch Operations for more details.

Update Multiple Objects

Update many objects in bulk:

# Get devices to update
devices = await client.filters(
    kind="InfraDevice",
    status__value="provisioning"
)

# Create batch
batch = await client.create_batch()

# Update each
for device in devices:
    device.status.value = "active"
    batch.add(task=device.save, node=device)

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

Delete Multiple Objects

Delete many objects:

# Get objects to delete
old_devices = await client.filters(
    kind="InfraDevice",
    status__value="obsolete"
)

# Delete each
for device in old_devices:
    await client.delete(node=device)
    print(f"Deleted: {device.name.value}")

Validation

Handle Validation Errors

Catch validation errors during save:

from infrahub_sdk.exceptions import GraphQLError

try:
    device = await client.create(
        kind="InfraDevice",
        name="",  # Invalid: empty name
        status="active"
    )
    await device.save()
except GraphQLError as e:
    print(f"Validation error: {e.message}")
    # Handle the error

Uniqueness Constraints

Handle uniqueness constraint violations:

try:
    device = await client.create(
        kind="InfraDevice",
        name="router-01",  # Name already exists
        status="active"
    )
    await device.save()
except GraphQLError as e:
    if "unique" in e.message.lower():
        print("Device with this name already exists")
    else:
        raise

Transaction Patterns

Rollback on Failure

Use branches for transaction-like behavior:

async def safe_multi_create():
    client = InfrahubClient()
    
    # Create a temporary branch
    branch = await client.branch.create(
        branch_name="temp-transaction",
        description="Temporary transaction branch"
    )
    
    try:
        # Create objects in the branch
        device1 = await client.create(
            branch="temp-transaction",
            kind="InfraDevice",
            name="router-01"
        )
        await device1.save()
        
        device2 = await client.create(
            branch="temp-transaction",
            kind="InfraDevice",
            name="router-02"
        )
        await device2.save()
        
        # If successful, merge to main
        await client.branch.merge(branch_name="temp-transaction")
        
    except Exception as e:
        # On error, delete the branch (rollback)
        await client.branch.delete(branch_name="temp-transaction")
        raise e

Examples

Create Site and Devices

async def setup_site(site_name: str, device_count: int):
    client = InfrahubClient()
    
    # Create site
    site = await client.create(
        kind="InfraSite",
        name=site_name,
        description=f"{site_name} datacenter"
    )
    await site.save()
    print(f"Created site: {site.name.value}")
    
    # Create devices
    batch = await client.create_batch()
    for i in range(1, device_count + 1):
        device = await client.create(
            kind="InfraDevice",
            name=f"{site_name.lower()}-router-{i:02d}",
            site=site,
            status="active"
        )
        batch.add(task=device.save, node=device)
    
    async for node, result in batch.execute():
        print(f"Created device: {node.name.value}")

# Usage
await setup_site("NYC", 5)

Update Device Status

async def set_maintenance_mode(device_name: str):
    client = InfrahubClient()
    
    device = await client.get(
        kind="InfraDevice",
        name__value=device_name
    )
    
    if device:
        # Update status
        device.status.value = "maintenance"
        
        # Add maintenance tag
        maint_tag = await client.get(
            kind="BuiltinTag",
            name__value="maintenance"
        )
        
        if maint_tag:
            current_tags = await device.tags.fetch()
            device.tags.peers = current_tags + [maint_tag]
        
        await device.save()
        print(f"Set {device_name} to maintenance mode")

# Usage
await set_maintenance_mode("router-01")

Decommission Device

async def decommission_device(device_id: str):
    client = InfrahubClient()
    
    device = await client.get(kind="InfraDevice", id=device_id)
    
    if device:
        # Update status
        device.status.value = "obsolete"
        await device.save()
        
        # Wait for confirmation
        print(f"Device {device.name.value} marked as obsolete")
        
        # Delete after marking obsolete
        await client.delete(node=device)
        print(f"Deleted device {device.name.value}")

# Usage
await decommission_device("abc-123")

Error Handling

Common Patterns

from infrahub_sdk.exceptions import (
    GraphQLError,
    ValidationError,
    Error
)

async def safe_create_device(name: str, **kwargs):
    client = InfrahubClient()
    
    try:
        device = await client.create(
            kind="InfraDevice",
            name=name,
            **kwargs
        )
        await device.save()
        return device
        
    except GraphQLError as e:
        if "unique" in e.message.lower():
            print(f"Device {name} already exists")
            # Return existing device
            return await client.get(
                kind="InfraDevice",
                name__value=name
            )
        else:
            print(f"GraphQL error: {e.message}")
            raise
            
    except Error as e:
        print(f"SDK error: {e.message}")
        raise

Documentation Index

​Creating Objects

​Basic Create

​Create with Relationships

​Create in a Branch

​Create with Protected Values

​Create with Source Tracking

​Updating Objects

​Basic Update

​Update Multiple Attributes

​Update Relationships

​Update in a Branch

​Protect Attributes

​Deleting Objects

​Basic Delete

​Delete by ID

​Delete in a Branch

​Working with Many Relationships

​Add to Many Relationship

​Update Many Relationship

​Remove from Many Relationship

​GraphQL Mutations

​Custom Mutations

​Upsert Pattern

​Bulk Operations

​Create Multiple Objects

​Update Multiple Objects

​Delete Multiple Objects

​Validation

​Handle Validation Errors

​Uniqueness Constraints

​Transaction Patterns

​Rollback on Failure

​Examples

​Create Site and Devices

​Update Device Status

​Decommission Device

​Error Handling

​Common Patterns

​Next Steps

Batch Operations

Queries

Creating Objects

Basic Create

Create with Relationships

Create in a Branch

Create with Protected Values

Create with Source Tracking

Updating Objects

Basic Update

Update Multiple Attributes

Update Relationships

Update in a Branch

Protect Attributes

Deleting Objects

Basic Delete

Delete by ID

Delete in a Branch

Working with Many Relationships

Add to Many Relationship

Update Many Relationship

Remove from Many Relationship

GraphQL Mutations

Custom Mutations

Upsert Pattern

Bulk Operations

Create Multiple Objects

Update Multiple Objects

Delete Multiple Objects

Validation

Handle Validation Errors

Uniqueness Constraints

Transaction Patterns

Rollback on Failure

Examples

Create Site and Devices

Update Device Status

Decommission Device

Error Handling

Common Patterns

Next Steps