Relationships

Manage relationships between infrastructure objects in the Infrahub Python SDK

Relationships connect objects in your infrastructure graph. The SDK provides intuitive methods to create, query, and manage these connections.

Understanding Relationships

Relationship Types

Infrahub supports different relationship types:

Attribute Relationships

Core relationships that define the structure of an object. These are required for the object’s identity.

Example: A Device’s location attribute

Generic Relationships

Optional connections between objects that don’t define the object’s core identity.

Example: A Device’s tags relationship

Cardinality

Relationships can have different cardinalities:

• One: Single related object (e.g., Device → Location)
• Many: Multiple related objects (e.g., Device → Tags)

Working with Single Relationships

Create with Relationship

Create an object with a single relationship:

from infrahub_sdk import InfrahubClient

client = InfrahubClient()

# Create location
location = await client.create(
    kind="InfraLocation",
    name="Datacenter-1"
)
await location.save()

# Create device with location relationship
device = await client.create(
    kind="InfraDevice",
    name="router-01",
    location=location  # Pass the object
)
await device.save()

Query with Relationship

Fetch an object and its related data:

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["location"]  # Include the relationship
)

print(f"Device: {device.name.value}")
print(f"Location: {device.location.name.value}")
print(f"Location ID: {device.location.id}")

Update Relationship

Change a single relationship:

device = await client.get(
    kind="InfraDevice",
    id="device-id"
)

# Get new location
new_location = await client.get(
    kind="InfraLocation",
    id="new-location-id"
)

# Update the relationship
device.location = new_location
await device.save()

Using Relationship IDs

# Create with ID
device = await client.create(
    kind="InfraDevice",
    name="router-02",
    location="location-id-123"  # Pass ID directly
)
await device.save()

# Update with ID
device.location = "new-location-id-456"
await device.save()

Access Relationship Properties

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["location"],
    property=True  # Include metadata
)

print(f"Location ID: {device.location.id}")
print(f"Updated at: {device.location.updated_at}")
print(f"From profile: {device.location.is_from_profile}")
print(f"Source: {device.location.source}")

Working with Many Relationships

Create with Multiple Relationships

Create an object with many related objects:

# Create 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 tags
device = await client.create(
    kind="InfraDevice",
    name="router-01",
    tags=[tag1, tag2]  # Pass list of objects
)
await device.save()

Query Many Relationships

Fetch objects with multiple related items:

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["tags"]  # Include many relationship
)

print(f"Device has {len(device.tags)} tags:")
for tag in device.tags:
    print(f"  - {tag.name.value}")

# Access tag IDs
tag_ids = device.tags.peer_ids
print(f"Tag IDs: {tag_ids}")

Add to Many Relationship

Add new items to an existing relationship:

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["tags"]
)

# Get existing tags
current_tags = list(device.tags)

# Create and add new tag
new_tag = await client.create(kind="BuiltinTag", name="new-tag")
await new_tag.save()

current_tags.append(new_tag)

# Update with all tags
device.tags = current_tags
await device.save()

print(f"Device now has {len(device.tags)} tags")

Remove from Many Relationship

Remove specific items:

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["tags"]
)

# Remove specific tag
tag_to_remove_id = "tag-id-123"
device.tags = [
    tag for tag in device.tags
    if tag.id != tag_to_remove_id
]

await device.save()

Replace All Related Objects

device = await client.get(
    kind="InfraDevice",
    id="device-id"
)

# Get new tags
new_tags = [
    await client.get(kind="BuiltinTag", id="tag-1"),
    await client.get(kind="BuiltinTag", id="tag-2")
]

# Replace all tags
device.tags = new_tags
await device.save()

Clear All Related Objects

device = await client.get(
    kind="InfraDevice",
    id="device-id"
)

# Clear all tags
device.tags = []
await device.save()

Nested Relationships

Query Nested Relationships

Fetch relationships of related objects:

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=[
        "location",
        "location__parent",  # Nested: location's parent
        "location__region"   # Nested: location's region
    ]
)

print(f"Device: {device.name.value}")
print(f"Location: {device.location.name.value}")

if device.location.parent:
    print(f"Parent Location: {device.location.parent.name.value}")

if device.location.region:
    print(f"Region: {device.location.region.name.value}")

Traverse Relationships

Navigate through multiple levels:

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["location", "location__parent", "location__parent__parent"]
)

# Traverse the hierarchy
current = device.location
level = 0

while current:
    print(f"{'  ' * level}Level {level}: {current.name.value}")
    current = current.parent if hasattr(current, "parent") else None
    level += 1

Bidirectional Relationships

Access Reverse Relationships

# Get location and all devices at that location
location = await client.get(
    kind="InfraLocation",
    id="location-id",
    include=["devices"]  # Reverse relationship
)

print(f"Location: {location.name.value}")
print(f"Devices at this location: {len(location.devices)}")

for device in location.devices:
    print(f"  - {device.name.value}")

Update from Both Sides

# Add device to location (forward)
device = await client.get(kind="InfraDevice", id="device-id")
location = await client.get(kind="InfraLocation", id="location-id")

device.location = location
await device.save()

# Verify from reverse side
location_updated = await client.get(
    kind="InfraLocation",
    id="location-id",
    include=["devices"]
)

device_ids = [d.id for d in location_updated.devices]
assert device.id in device_ids

Relationship Filters

Filter Related Objects

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["tags"]
)

# Filter tags by name
production_tags = [
    tag for tag in device.tags
    if "prod" in tag.name.value.lower()
]

print(f"Production tags: {[t.name.value for t in production_tags]}")

Count Related Objects

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["interfaces", "tags"]
)

interface_count = len(device.interfaces)
tag_count = len(device.tags)

print(f"Interfaces: {interface_count}")
print(f"Tags: {tag_count}")

Check for Specific Relationship

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["tags"]
)

# Check if device has specific tag
target_tag_id = "production-tag-id"
has_tag = any(tag.id == target_tag_id for tag in device.tags)

if has_tag:
    print("Device is tagged as production")
else:
    print("Device is not in production")

Relationship Validation

Validate Relationships Exist

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["location"]
)

if not device.location:
    print("Warning: Device has no location assigned")
else:
    print(f"Device is at: {device.location.name.value}")

Validate Relationship Constraints

device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["tags"]
)

# Ensure minimum number of tags
min_tags = 2
if len(device.tags) < min_tags:
    print(f"Error: Device must have at least {min_tags} tags")
    # Add required tags...

Bulk Relationship Operations

Update Relationships for Multiple Objects

# Get all devices
devices = await client.all(kind="InfraDevice")

# Get new location
new_location = await client.get(kind="InfraLocation", id="new-location-id")

# Update location for all devices
for device in devices:
    device.location = new_location
    await device.save()

print(f"Updated location for {len(devices)} devices")

Add Tag to Multiple Objects

# Create tag
prod_tag = await client.create(kind="BuiltinTag", name="production")
await prod_tag.save()

# Get all devices
devices = await client.all(kind="InfraDevice")

# Add tag to all devices
for device in devices:
    device_full = await client.get(
        kind="InfraDevice",
        id=device.id,
        include=["tags"]
    )
    
    current_tags = list(device_full.tags)
    current_tags.append(prod_tag)
    device_full.tags = current_tags
    
    await device_full.save()

print(f"Tagged {len(devices)} devices")

Batch Relationship Updates

tag_updates = {
    "device-1": ["tag-a", "tag-b"],
    "device-2": ["tag-a", "tag-c"],
    "device-3": ["tag-b", "tag-c"]
}

for device_id, tag_ids in tag_updates.items():
    device = await client.get(
        kind="InfraDevice",
        id=device_id
    )
    
    device.tags = tag_ids  # Assign IDs
    await device.save()

print(f"Updated tags for {len(tag_updates)} devices")

Advanced Patterns

Relationship Builder

class DeviceRelationshipManager:
    def __init__(self, device):
        self.device = device
        self._tags = None
    
    async def load_tags(self):
        """Lazy load tags."""
        if self._tags is None:
            device_full = await client.get(
                kind="InfraDevice",
                id=self.device.id,
                include=["tags"]
            )
            self._tags = list(device_full.tags)
        return self._tags
    
    async def add_tag(self, tag):
        """Add a tag to the device."""
        tags = await self.load_tags()
        
        if tag.id not in [t.id for t in tags]:
            tags.append(tag)
            self.device.tags = tags
            await self.device.save()
    
    async def remove_tag(self, tag_id):
        """Remove a tag from the device."""
        tags = await self.load_tags()
        
        self.device.tags = [t for t in tags if t.id != tag_id]
        await self.device.save()

# Usage
device = await client.get(kind="InfraDevice", id="device-id")
manager = DeviceRelationshipManager(device)

new_tag = await client.create(kind="BuiltinTag", name="critical")
await new_tag.save()

await manager.add_tag(new_tag)

Relationship Graph Traversal

async def find_all_related(
    client: InfrahubClient,
    start_kind: str,
    start_id: str,
    relationship: str,
    max_depth: int = 5
) -> list:
    """Find all related objects recursively."""
    visited = set()
    to_visit = [(start_kind, start_id, 0)]
    all_related = []
    
    while to_visit:
        kind, obj_id, depth = to_visit.pop(0)
        
        if obj_id in visited or depth > max_depth:
            continue
        
        visited.add(obj_id)
        
        obj = await client.get(
            kind=kind,
            id=obj_id,
            include=[relationship]
        )
        
        related = getattr(obj, relationship, None)
        if related:
            if isinstance(related, list):
                for rel in related:
                    all_related.append(rel)
                    to_visit.append((rel._schema.kind, rel.id, depth + 1))
            else:
                all_related.append(related)
                to_visit.append((related._schema.kind, related.id, depth + 1))
    
    return all_related

# Find all locations in hierarchy
locations = await find_all_related(
    client=client,
    start_kind="InfraLocation",
    start_id="root-location-id",
    relationship="children",
    max_depth=10
)

Next Steps

Branches

Work with Git-like branches

Batch Operations

Perform efficient bulk operations

Error Handling

Handle relationship errors

Async Operations

Use async patterns effectively