Updating Objects - Infrahub

The SDK provides intuitive methods to update objects and their attributes in your infrastructure graph.

Basic Updates

Update and Save Pattern

The standard workflow for updating objects:

from infrahub_sdk import InfrahubClient

client = InfrahubClient()

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

# Update attribute values
device.name.value = "router-updated"
device.description.value = "Updated description"

# Save changes
await device.save()

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

Changes are not persisted until you call .save(). This allows you to make multiple changes before committing.

Update Single Attribute

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

# Update just one attribute
device.is_active.value = False
await device.save()

Update Multiple Attributes

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

# Update multiple attributes at once
device.name.value = "new-name"
device.serial_number.value = "SN999999"
device.height.value = 42
device.is_active.value = True

await device.save()

Updating Different Attribute Types

Text Attributes

device.name.value = "new-name"
device.description.value = "Updated description"
await device.save()

Number Attributes

device.height.value = 42  # Integer
device.weight.value = 25.5  # Float
await device.save()

Boolean Attributes

device.is_active.value = False
device.is_managed.value = True
await device.save()

DateTime Attributes

from datetime import datetime

device.commissioned_at.value = datetime.now()
await device.save()

JSON Attributes

device.metadata.value = {
    "vendor": "Cisco",
    "model": "ISR4451",
    "firmware": "17.9.1"
}
await device.save()

Setting Attributes to None

# Clear optional attributes
device.description.value = None
device.height.value = None
await device.save()

Updating Relationships

Update Single Cardinality Relationships

Change one-to-one relationships:

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

# Get new location
new_location = await client.get(kind="InfraLocation", id="new-location-id")

# Update the relationship
device.location = new_location
await device.save()

print(f"Device moved to: {new_location.name.value}")

Using Relationship IDs

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

# Update using ID instead of object
device.location = "new-location-id"
await device.save()

Update Many-to-Many Relationships

Manage relationships with multiple objects:

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

# Replace all tags
new_tags = [
    await client.get(kind="BuiltinTag", id="tag-1"),
    await client.get(kind="BuiltinTag", id="tag-2")
]
device.tags = new_tags
await device.save()

Add to Many-to-Many Relationships

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

# Get existing tags
current_tags = list(device.tags)

# Add new tag
new_tag = await client.get(kind="BuiltinTag", id="new-tag-id")
current_tags.append(new_tag)

# Update with combined list
device.tags = current_tags
await device.save()

Remove from Many-to-Many Relationships

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

# Filter out specific tag
device.tags = [
    tag for tag in device.tags
    if tag.id != "tag-to-remove-id"
]
await device.save()

Clear All Relationships

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

# Clear many-to-many relationship
device.tags = []

# Clear optional one-to-one relationship
device.location = None

await device.save()

Updating on Branches

Update on Specific Branch

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

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

# Make changes
device.name.value = "updated-name"
await device.save()

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

Compare Before and After

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

# Create branch and update
branch = await client.branch.create(branch_name="feature-update")
branch_device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch=branch.name
)

branch_device.name.value = "new-name"
await branch_device.save()

print(f"Main: {main_device.name.value}")
print(f"Branch: {branch_device.name.value}")

Batch Updates

Update Multiple Objects

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

# Update all devices
for device in devices:
    device.is_managed.value = True
    await device.save()

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

Conditional Bulk Updates

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

# Update only inactive devices
updated_count = 0
for device in devices:
    if not device.is_active.value:
        device.is_active.value = True
        await device.save()
        updated_count += 1

print(f"Activated {updated_count} devices")

Update with Filtering

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

# Update devices matching criteria
for device in devices:
    if "router" in device.name.value.lower():
        device.device_type.value = "router"
        await device.save()

Partial Updates

Update Only Changed Fields

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

# Only update what changed
if device.name.value != "desired-name":
    device.name.value = "desired-name"
    await device.save()

Merge Updates

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

updates = {
    "name": "new-name",
    "is_active": True,
    "height": 42
}

# Apply updates
for attr, value in updates.items():
    if hasattr(device, attr):
        getattr(device, attr).value = value

await device.save()

Validation and Error Handling

Handle Update Errors

from infrahub_sdk.exceptions import GraphQLError

try:
    device = await client.get(kind="InfraDevice", id="device-id")
    device.name.value = ""  # Invalid: empty name
    await device.save()
    
except GraphQLError as e:
    print(f"Update failed: {e.message}")

Validate Before Updating

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

new_name = "router-01"
if len(new_name) >= 3:  # Validation rule
    device.name.value = new_name
    await device.save()
else:
    print("Invalid name: too short")

Handle Constraint Violations

from infrahub_sdk.exceptions import GraphQLError

try:
    device = await client.get(kind="InfraDevice", id="device-id")
    device.serial_number.value = "duplicate-serial"  # May violate uniqueness
    await device.save()
    
except GraphQLError as e:
    if "unique" in e.message.lower():
        print("Serial number must be unique")
    else:
        print(f"Update error: {e.message}")

Advanced Update Patterns

Atomic Updates

async def atomic_update(
    client: InfrahubClient,
    device_id: str,
    updates: dict
) -> bool:
    """Perform atomic update with rollback on error."""
    device = await client.get(kind="InfraDevice", id=device_id)
    
    # Store original values
    original_values = {
        attr: getattr(device, attr).value
        for attr in updates.keys()
    }
    
    try:
        # Apply updates
        for attr, value in updates.items():
            getattr(device, attr).value = value
        
        await device.save()
        return True
        
    except Exception as e:
        # Rollback on error
        for attr, value in original_values.items():
            getattr(device, attr).value = value
        
        print(f"Update failed, rolled back: {e}")
        return False

# Use the function
success = await atomic_update(
    client=client,
    device_id="device-id",
    updates={"name": "new-name", "is_active": True}
)

Conditional Updates

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

# Update only if condition is met
if device.is_active.value:
    device.last_check.value = datetime.now()
    await device.save()

Update with Retry

import asyncio
from infrahub_sdk.exceptions import GraphQLError

async def update_with_retry(
    client: InfrahubClient,
    device_id: str,
    updates: dict,
    max_retries: int = 3
) -> bool:
    """Update with automatic retry on failure."""
    for attempt in range(max_retries):
        try:
            device = await client.get(kind="InfraDevice", id=device_id)
            
            for attr, value in updates.items():
                getattr(device, attr).value = value
            
            await device.save()
            return True
            
        except GraphQLError as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)  # Exponential backoff
                continue
            else:
                print(f"Update failed after {max_retries} attempts: {e}")
                return False
    
    return False

Track Changes

class ChangeTracker:
    def __init__(self, obj):
        self.obj = obj
        self.changes = {}
        self.original_values = {}
    
    def update(self, attr: str, value):
        if attr not in self.original_values:
            self.original_values[attr] = getattr(self.obj, attr).value
        
        getattr(self.obj, attr).value = value
        self.changes[attr] = {
            "old": self.original_values[attr],
            "new": value
        }
    
    async def save(self):
        await self.obj.save()
        return self.changes

# Use the tracker
device = await client.get(kind="InfraDevice", id="device-id")
tracker = ChangeTracker(device)

tracker.update("name", "new-name")
tracker.update("height", 42)

changes = await tracker.save()
for attr, change in changes.items():
    print(f"{attr}: {change['old']} -> {change['new']}")

Update Strategies

Optimistic Updates

# Assume update will succeed, handle errors after
device = await client.get(kind="InfraDevice", id="device-id")

device.name.value = "new-name"
print(f"Updated to: {device.name.value}")  # Show immediately

try:
    await device.save()
except Exception as e:
    print(f"Update failed: {e}")
    # Handle rollback or user notification

Pessimistic Updates

# Validate before attempting update
device = await client.get(kind="InfraDevice", id="device-id")

new_name = "new-name"

# Validate first
if not new_name:
    print("Invalid name")
else:
    device.name.value = new_name
    await device.save()
    print(f"Updated to: {device.name.value}")

Incremental Updates

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

# Update incrementally
device.name.value = "step-1"
await device.save()

device.height.value = 42
await device.save()

device.is_active.value = True
await device.save()

Next Steps

Deleting Objects

Learn how to delete objects

Relationships

Master relationship management

Batch Operations

Perform efficient bulk operations

Error Handling

Handle errors gracefully

Documentation Index

​Basic Updates

​Update and Save Pattern

​Update Single Attribute

​Update Multiple Attributes

​Updating Different Attribute Types

​Text Attributes

​Number Attributes

​Boolean Attributes

​DateTime Attributes

​JSON Attributes

​Setting Attributes to None

​Updating Relationships

​Update Single Cardinality Relationships

​Using Relationship IDs

​Update Many-to-Many Relationships

​Add to Many-to-Many Relationships

​Remove from Many-to-Many Relationships

​Clear All Relationships

​Updating on Branches

​Update on Specific Branch

​Compare Before and After

​Batch Updates

​Update Multiple Objects

​Conditional Bulk Updates

​Update with Filtering

​Partial Updates

​Update Only Changed Fields

​Merge Updates

​Validation and Error Handling

​Handle Update Errors

​Validate Before Updating

​Handle Constraint Violations

​Advanced Update Patterns

​Atomic Updates

​Conditional Updates

​Update with Retry

​Track Changes

​Update Strategies

​Optimistic Updates

​Pessimistic Updates

​Incremental Updates

​Next Steps

Deleting Objects

Relationships

Batch Operations

Error Handling

Basic Updates

Update and Save Pattern

Update Single Attribute

Update Multiple Attributes

Updating Different Attribute Types

Text Attributes

Number Attributes

Boolean Attributes

DateTime Attributes

JSON Attributes

Setting Attributes to None

Updating Relationships

Update Single Cardinality Relationships

Using Relationship IDs

Update Many-to-Many Relationships

Add to Many-to-Many Relationships

Remove from Many-to-Many Relationships

Clear All Relationships

Updating on Branches

Update on Specific Branch

Compare Before and After

Batch Updates

Update Multiple Objects

Conditional Bulk Updates

Update with Filtering

Partial Updates

Update Only Changed Fields

Merge Updates

Validation and Error Handling

Handle Update Errors

Validate Before Updating

Handle Constraint Violations

Advanced Update Patterns

Atomic Updates

Conditional Updates

Update with Retry

Track Changes

Update Strategies

Optimistic Updates

Pessimistic Updates

Incremental Updates

Next Steps