Skip to main content
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

Build docs developers (and LLMs) love