Deleting Objects - Infrahub

The SDK provides methods to safely delete objects from your infrastructure graph.

Basic Deletion

Delete an Object

The standard pattern for deleting objects:

from infrahub_sdk import InfrahubClient

client = InfrahubClient()

# Get the object
device = await client.get(kind="InfraDevice", id="device-id")

# Delete it
await device.delete()

print(f"Deleted device: {device.name.value}")

Deletion is permanent and cannot be undone. Always verify before deleting.

Delete by ID

# Delete without fetching first
tag = await client.get(kind="BuiltinTag", id="tag-id")
await tag.delete()

Verifying Before Deletion

Confirm Before Delete

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

print(f"About to delete: {device.name.value}")
confirm = input("Are you sure? (yes/no): ")

if confirm.lower() == "yes":
    await device.delete()
    print("Device deleted")
else:
    print("Deletion cancelled")

Check Relationships Before Delete

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

# Check for dependencies
if device.interfaces or device.connections:
    print("Cannot delete: device has dependencies")
    print(f"  Interfaces: {len(device.interfaces)}")
    print(f"  Connections: {len(device.connections)}")
else:
    await device.delete()
    print("Device deleted successfully")

Cascade Deletion

Delete an object and its related components:

# Get device with all interfaces
device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["interfaces"]
)

# Delete all interfaces first
for interface in device.interfaces:
    await interface.delete()
    print(f"Deleted interface: {interface.name.value}")

# Then delete the device
await device.delete()
print("Device and all interfaces deleted")

Delete Hierarchy

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

# Delete all devices
for device in location.devices:
    await device.delete()

# Delete the location
await location.delete()
print(f"Deleted location and {len(location.devices)} devices")

Batch Deletion

Delete Multiple Objects

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

# Delete devices matching criteria
deleted_count = 0
for device in devices:
    if not device.is_active.value:
        await device.delete()
        deleted_count += 1

print(f"Deleted {deleted_count} inactive devices")

Delete by Filter

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

# Delete test devices
for device in devices:
    if device.name.value.startswith("test-"):
        await device.delete()
        print(f"Deleted: {device.name.value}")

Delete All Objects of a Kind

This operation deletes all objects of the specified kind. Use with extreme caution.

# Delete all tags
tags = await client.all(kind="BuiltinTag")

confirm = input(f"Delete all {len(tags)} tags? (yes/no): ")
if confirm.lower() == "yes":
    for tag in tags:
        await tag.delete()
    print(f"Deleted {len(tags)} tags")

Deleting on Branches

Delete on Specific Branch

# Create a branch
branch = await client.branch.create(branch_name="cleanup")

# Get and delete object on that branch
device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch=branch.name
)

await device.delete()
print(f"Deleted device on branch: {branch.name}")

Test Deletion on Branch

# Create test branch
test_branch = await client.branch.create(branch_name="test-deletion")

# Delete on test branch
device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch=test_branch.name
)
await device.delete()

# Verify on main branch (still exists)
main_device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch="main"
)
print(f"Device still exists on main: {main_device.name.value}")

Error Handling

Handle Deletion Errors

from infrahub_sdk.exceptions import GraphQLError, NodeNotFoundError

try:
    device = await client.get(kind="InfraDevice", id="device-id")
    await device.delete()
    print("Device deleted successfully")
    
except NodeNotFoundError:
    print("Device not found")
except GraphQLError as e:
    print(f"Deletion failed: {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

Handle Constraint Violations

from infrahub_sdk.exceptions import GraphQLError

try:
    location = await client.get(kind="InfraLocation", id="location-id")
    await location.delete()
    
except GraphQLError as e:
    if "constraint" in e.message.lower():
        print("Cannot delete: location has dependent devices")
        print("Delete devices first, then try again")
    else:
        print(f"Deletion error: {e.message}")

Safe Deletion with Retry

import asyncio
from infrahub_sdk.exceptions import GraphQLError

async def safe_delete(
    client: InfrahubClient,
    kind: str,
    object_id: str,
    max_retries: int = 3
) -> bool:
    """Delete with retry on transient errors."""
    for attempt in range(max_retries):
        try:
            obj = await client.get(kind=kind, id=object_id)
            await obj.delete()
            return True
            
        except GraphQLError as e:
            if "constraint" in str(e).lower():
                # Don't retry constraint violations
                print(f"Cannot delete: {e.message}")
                return False
            elif attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
                continue
            else:
                print(f"Deletion failed after {max_retries} attempts")
                return False
    
    return False

# Use the function
success = await safe_delete(
    client=client,
    kind="InfraDevice",
    object_id="device-id"
)

Soft Deletion Pattern

Mark as Deleted Instead of Removing

# Instead of deleting, mark as inactive
device = await client.get(kind="InfraDevice", id="device-id")

device.is_active.value = False
device.deleted_at.value = datetime.now()
await device.save()

print(f"Soft-deleted device: {device.name.value}")

Cleanup Soft-Deleted Objects

from datetime import datetime, timedelta

# Delete objects marked as deleted over 30 days ago
devices = await client.all(kind="InfraDevice")

cutoff_date = datetime.now() - timedelta(days=30)
deleted_count = 0

for device in devices:
    if (
        not device.is_active.value and
        device.deleted_at.value and
        device.deleted_at.value < cutoff_date
    ):
        await device.delete()
        deleted_count += 1

print(f"Cleaned up {deleted_count} soft-deleted devices")

Advanced Deletion Patterns

Dry Run Mode

async def delete_with_dry_run(
    client: InfrahubClient,
    kind: str,
    filter_func,
    dry_run: bool = True
):
    """Preview deletions before executing."""
    objects = await client.all(kind=kind)
    to_delete = [obj for obj in objects if filter_func(obj)]
    
    if dry_run:
        print(f"Would delete {len(to_delete)} objects:")
        for obj in to_delete:
            print(f"  - {obj.id}")
        return to_delete
    else:
        for obj in to_delete:
            await obj.delete()
        print(f"Deleted {len(to_delete)} objects")
        return to_delete

# Preview
await delete_with_dry_run(
    client=client,
    kind="InfraDevice",
    filter_func=lambda d: not d.is_active.value,
    dry_run=True
)

# Execute
await delete_with_dry_run(
    client=client,
    kind="InfraDevice",
    filter_func=lambda d: not d.is_active.value,
    dry_run=False
)

Track Deletions

class DeletionTracker:
    def __init__(self):
        self.deleted = []
        self.failed = []
    
    async def delete(self, obj):
        try:
            obj_info = {
                "id": obj.id,
                "kind": obj._schema.kind,
                "name": obj.name.value if hasattr(obj, "name") else None
            }
            
            await obj.delete()
            self.deleted.append(obj_info)
            return True
            
        except Exception as e:
            self.failed.append({
                **obj_info,
                "error": str(e)
            })
            return False
    
    def report(self):
        print(f"Deleted: {len(self.deleted)}")
        print(f"Failed: {len(self.failed)}")
        
        if self.failed:
            print("\nFailed deletions:")
            for item in self.failed:
                print(f"  - {item['id']}: {item['error']}")

# Use the tracker
tracker = DeletionTracker()

devices = await client.all(kind="InfraDevice")
for device in devices:
    if not device.is_active.value:
        await tracker.delete(device)

tracker.report()

Atomic Batch Deletion

async def atomic_batch_delete(
    client: InfrahubClient,
    kind: str,
    object_ids: list[str]
) -> bool:
    """Delete multiple objects, rollback on any failure."""
    deleted = []
    
    try:
        for object_id in object_ids:
            obj = await client.get(kind=kind, id=object_id)
            await obj.delete()
            deleted.append(obj)
        
        print(f"Deleted {len(deleted)} objects")
        return True
        
    except Exception as e:
        print(f"Error during deletion: {e}")
        print("Note: Some objects may have been deleted")
        # In a real scenario, you might need to recreate deleted objects
        return False

# Use the function
success = await atomic_batch_delete(
    client=client,
    kind="InfraDevice",
    object_ids=["id-1", "id-2", "id-3"]
)

Cleanup Utilities

Delete Orphaned Objects

# Find and delete devices without a location
devices = await client.all(kind="InfraDevice")

orphaned = []
for device in devices:
    device_full = await client.get(
        kind="InfraDevice",
        id=device.id,
        include=["location"]
    )
    
    if not device_full.location:
        orphaned.append(device_full)

print(f"Found {len(orphaned)} orphaned devices")

for device in orphaned:
    await device.delete()
    print(f"Deleted orphaned device: {device.name.value}")

Delete Duplicates

# Find and delete duplicate devices (by serial number)
devices = await client.all(kind="InfraDevice")

seen = {}
duplicates = []

for device in devices:
    serial = device.serial_number.value
    
    if serial in seen:
        duplicates.append(device)
    else:
        seen[serial] = device

print(f"Found {len(duplicates)} duplicate devices")

for device in duplicates:
    await device.delete()
    print(f"Deleted duplicate: {device.name.value}")

Next Steps

Relationships

Understand relationship management

Batch Operations

Perform efficient bulk operations

Error Handling

Handle errors and exceptions

Branches

Work with Git-like branches

Documentation Index

​Basic Deletion

​Delete an Object

​Delete by ID

​Verifying Before Deletion

​Confirm Before Delete

​Check Relationships Before Delete

​Cascade Deletion

​Delete Related Objects

​Delete Hierarchy

​Batch Deletion

​Delete Multiple Objects

​Delete by Filter

​Delete All Objects of a Kind

​Deleting on Branches

​Delete on Specific Branch

​Test Deletion on Branch

​Error Handling

​Handle Deletion Errors

​Handle Constraint Violations

​Safe Deletion with Retry

​Soft Deletion Pattern

​Mark as Deleted Instead of Removing

​Cleanup Soft-Deleted Objects

​Advanced Deletion Patterns

​Dry Run Mode

​Track Deletions

​Atomic Batch Deletion

​Cleanup Utilities

​Delete Orphaned Objects

​Delete Duplicates

​Next Steps

Relationships

Batch Operations

Error Handling

Branches

Basic Deletion

Delete an Object

Delete by ID

Verifying Before Deletion

Confirm Before Delete

Check Relationships Before Delete

Cascade Deletion

Delete Related Objects

Delete Hierarchy

Batch Deletion

Delete Multiple Objects

Delete by Filter

Delete All Objects of a Kind

Deleting on Branches

Delete on Specific Branch

Test Deletion on Branch

Error Handling

Handle Deletion Errors

Handle Constraint Violations

Safe Deletion with Retry

Soft Deletion Pattern

Mark as Deleted Instead of Removing

Cleanup Soft-Deleted Objects

Advanced Deletion Patterns

Dry Run Mode

Track Deletions

Atomic Batch Deletion

Cleanup Utilities

Delete Orphaned Objects

Delete Duplicates

Next Steps