The SDK provides flexible methods to create new objects in your infrastructure graph.
Basic Object Creation
Create and Save
The standard pattern for creating objects:
from infrahub_sdk import InfrahubClient
client = InfrahubClient()
# Create a new tag
tag = await client.create(
kind = "BuiltinTag" ,
name = "production" ,
description = "Production environment"
)
# Save to Infrahub
await tag.save()
print ( f "Created tag with ID: { tag.id } " )
Objects created with client.create() exist only in memory until you call .save().
Using the data Parameter
Pass attributes as a dictionary:
data = {
"name" : "production" ,
"description" : "Production environment"
}
tag = await client.create( kind = "BuiltinTag" , data = data)
await tag.save()
Mixing Parameters and Data
# Both styles work
tag = await client.create(
kind = "BuiltinTag" ,
data = { "name" : "staging" },
description = "Staging environment" # Direct parameter
)
await tag.save()
Creating Objects with Relationships
Single Cardinality Relationships
Create objects with relationships to other objects:
# First, create or get the related object
location = await client.create(
kind = "InfraLocation" ,
name = "Datacenter-1" ,
address = "123 Main St"
)
await location.save()
# Create a device with a relationship to the location
device = await client.create(
kind = "InfraDevice" ,
name = "router-01" ,
serial_number = "SN123456" ,
location = location # Pass the object directly
)
await device.save()
print ( f "Created device at location: { location.name.value } " )
Using Object IDs
Pass IDs instead of objects:
# Get or create location
location = await client.get( kind = "InfraLocation" , id = "location-id" )
# Create device using location ID
device = await client.create(
kind = "InfraDevice" ,
name = "router-02" ,
serial_number = "SN123457" ,
location = location.id # Pass the ID
)
await device.save()
Multiple Cardinality Relationships
Create objects with many-to-many relationships:
# Create multiple tags
tag1 = await client.create( kind = "BuiltinTag" , name = "production" )
await tag1.save()
tag2 = await client.create( kind = "BuiltinTag" , name = "critical" )
await tag2.save()
# Create device with multiple tags
device = await client.create(
kind = "InfraDevice" ,
name = "router-03" ,
serial_number = "SN123458" ,
tags = [tag1, tag2] # Pass list of objects
)
await device.save()
print ( f "Created device with { len (device.tags) } tags" )
Using Relationship IDs
tag_ids = [ "tag-id-1" , "tag-id-2" , "tag-id-3" ]
device = await client.create(
kind = "InfraDevice" ,
name = "router-04" ,
serial_number = "SN123459" ,
tags = tag_ids # Pass list of IDs
)
await device.save()
Attribute Types
Text Attributes
device = await client.create(
kind = "InfraDevice" ,
name = "router-01" , # Required text
description = "Main datacenter router" # Optional text
)
await device.save()
Number Attributes
device = await client.create(
kind = "InfraDevice" ,
name = "server-01" ,
height = 42 , # Rack units
weight = 25.5 # Kilograms (float)
)
await device.save()
Boolean Attributes
device = await client.create(
kind = "InfraDevice" ,
name = "router-01" ,
is_active = True ,
is_managed = False
)
await device.save()
DateTime Attributes
from datetime import datetime
device = await client.create(
kind = "InfraDevice" ,
name = "router-01" ,
commissioned_at = datetime( 2024 , 1 , 15 , 10 , 30 , 0 )
)
await device.save()
JSON Attributes
metadata = {
"vendor" : "Cisco" ,
"model" : "ISR4451" ,
"specs" : {
"cpu" : "4 cores" ,
"memory" : "8GB"
}
}
device = await client.create(
kind = "InfraDevice" ,
name = "router-01" ,
metadata = metadata
)
await device.save()
IP Address Attributes
interface = await client.create(
kind = "InfraInterface" ,
name = "eth0" ,
ip_address = "192.168.1.1" ,
netmask = "255.255.255.0"
)
await interface.save()
Creating on Branches
Create on a Specific Branch
# Create a new branch
branch = await client.branch.create( branch_name = "feature-update" )
# Create object on that branch
device = await client.create(
kind = "InfraDevice" ,
name = "router-new" ,
serial_number = "SN999999" ,
branch = branch.name # Specify the branch
)
await device.save()
print ( f "Created device on branch: { branch.name } " )
Workflow with Branches
# Create feature branch
feature_branch = await client.branch.create( branch_name = "add-devices" )
# Create multiple devices on the branch
for i in range ( 5 ):
device = await client.create(
kind = "InfraDevice" ,
name = f "device- { i :02d} " ,
serial_number = f "SN { i :06d} " ,
branch = feature_branch.name
)
await device.save()
print ( f "Created 5 devices on branch ' { feature_branch.name } '" )
Batch Creation
Create Multiple Objects
# Create multiple devices
devices = []
for i in range ( 10 ):
device = await client.create(
kind = "InfraDevice" ,
name = f "device- { i :02d} " ,
serial_number = f "SN { i :06d} "
)
await device.save()
devices.append(device)
print ( f "Created { len (devices) } devices" )
With Shared Relationships
# Create shared location
location = await client.create(
kind = "InfraLocation" ,
name = "Datacenter-1"
)
await location.save()
# Create multiple devices at the same location
devices = []
for i in range ( 10 ):
device = await client.create(
kind = "InfraDevice" ,
name = f "device- { i :02d} " ,
serial_number = f "SN { i :06d} " ,
location = location # Shared relationship
)
await device.save()
devices.append(device)
Validation and Constraints
Required Attributes
Ensure required attributes are provided:
try :
# Missing required 'name' attribute
device = await client.create(
kind = "InfraDevice" ,
serial_number = "SN123456" # name is missing
)
await device.save()
except Exception as e:
print ( f "Validation error: { e } " )
Unique Constraints
Handle unique constraint violations:
from infrahub_sdk.exceptions import GraphQLError
try :
# Create device with duplicate name (if name is unique)
device1 = await client.create(
kind = "InfraDevice" ,
name = "router-01" ,
serial_number = "SN111111"
)
await device1.save()
# This will fail if name must be unique
device2 = await client.create(
kind = "InfraDevice" ,
name = "router-01" , # Duplicate name
serial_number = "SN222222"
)
await device2.save()
except GraphQLError as e:
print ( f "Constraint violation: { e.message } " )
Custom Validation
Validate data before creation:
import re
def validate_serial_number ( serial : str ) -> bool :
"""Validate serial number format."""
pattern = r ' ^ SN \d {6} $ '
return bool (re.match(pattern, serial))
serial = "SN123456"
if validate_serial_number(serial):
device = await client.create(
kind = "InfraDevice" ,
name = "router-01" ,
serial_number = serial
)
await device.save()
else :
print ( "Invalid serial number format" )
Error Handling
Handle Creation Errors
from infrahub_sdk.exceptions import GraphQLError
try :
device = await client.create(
kind = "InfraDevice" ,
name = "router-01" ,
serial_number = "SN123456" ,
location = "invalid-id" # Invalid location ID
)
await device.save()
except GraphQLError as e:
print ( f "Failed to create device: { e.message } " )
# Handle the error appropriately
except Exception as e:
print ( f "Unexpected error: { e } " )
Rollback on Error
created_objects = []
try :
for i in range ( 10 ):
device = await client.create(
kind = "InfraDevice" ,
name = f "device- { i :02d} " ,
serial_number = f "SN { i :06d} "
)
await device.save()
created_objects.append(device)
except Exception as e:
print ( f "Error creating devices: { e } " )
# Rollback: delete created objects
for obj in created_objects:
await obj.delete()
print ( f "Rolled back { len (created_objects) } objects" )
Advanced Patterns
Factory Pattern
async def create_device (
client : InfrahubClient,
name : str ,
serial : str ,
location_id : str ,
** kwargs
):
"""Factory function for creating devices."""
device = await client.create(
kind = "InfraDevice" ,
name = name,
serial_number = serial,
location = location_id,
** kwargs
)
await device.save()
return device
# Use the factory
device = await create_device(
client = client,
name = "router-01" ,
serial = "SN123456" ,
location_id = "loc-123" ,
is_active = True
)
Builder Pattern
class DeviceBuilder :
def __init__ ( self , client : InfrahubClient):
self .client = client
self .data = {}
def with_name ( self , name : str ):
self .data[ "name" ] = name
return self
def with_serial ( self , serial : str ):
self .data[ "serial_number" ] = serial
return self
def at_location ( self , location_id : str ):
self .data[ "location" ] = location_id
return self
async def build ( self ):
device = await self .client.create(
kind = "InfraDevice" ,
** self .data
)
await device.save()
return device
# Use the builder
device = await (
DeviceBuilder(client)
.with_name( "router-01" )
.with_serial( "SN123456" )
.at_location( "loc-123" )
.build()
)
Template-Based Creation
device_template = {
"kind" : "InfraDevice" ,
"is_active" : True ,
"is_managed" : True ,
"location" : "datacenter-1"
}
# Create devices from template
for i in range ( 5 ):
device_data = device_template.copy()
device_data[ "name" ] = f "device- { i :02d} "
device_data[ "serial_number" ] = f "SN { i :06d} "
device = await client.create( ** device_data)
await device.save()
Next Steps
Updating Objects Learn how to update existing objects
Relationships Master working with relationships
Batch Operations Perform bulk operations efficiently
Error Handling Handle errors and exceptions properly