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