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
Next Steps
Batch Operations Learn efficient bulk operations
Queries Query data from Infrahub