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 Related Objects
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