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