Querying Data

Query and retrieve infrastructure objects from Infrahub using the Python SDK

The Infrahub Python SDK provides powerful methods to query and retrieve infrastructure data.

Basic Queries

Get a Single Object by ID

Retrieve an object using its unique identifier:

from infrahub_sdk import InfrahubClient

client = InfrahubClient()

device = await client.get(
    kind="InfraDevice",
    id="abc-123-def-456"
)

print(f"Device: {device.name.value}")
print(f"Serial: {device.serial_number.value}")

Get All Objects of a Kind

Retrieve all objects of a specific type:

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

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

Get with Specific Attributes

Retrieve an object with specific attribute values:

# Get by name (if unique)
device = await client.get(
    kind="InfraDevice",
    name__value="router-01"  # Use attribute filters
)

Including Relationships

Include Related Objects

Fetch related data in a single query:

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

# Access related data
print(f"Location: {device.location.name.value}")
print(f"Tags: {[tag.name.value for tag in device.tags]}")

Include Nested Relationships

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

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

Include All Relationships

# Include all direct relationships
device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include="all"
)

Property Mode

Fetch with Property Metadata

Retrieve additional property information:

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

# Access property metadata
print(f"Name: {device.name.value}")
print(f"Name updated at: {device.name.updated_at}")
print(f"Name is from profile: {device.name.is_from_profile}")
print(f"Name source: {device.name.source}")

Property mode includes metadata like:

• updated_at: When the attribute was last modified
• is_from_profile: Whether the value comes from a profile
• source: The source of the attribute value
• owner: Who owns/set the value

Filtering Results

Filter by Attribute Values

While the SDK doesn’t provide built-in filtering on all(), you can filter in Python:

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

# Filter by attribute
routers = [
    device for device in all_devices
    if device.device_type.value == "router"
]

# Filter by name pattern
prod_devices = [
    device for device in all_devices
    if "prod" in device.name.value.lower()
]

Custom GraphQL Queries

For complex filtering, use custom GraphQL queries:

query = """
query GetDevicesByType($device_type: String!) {
  InfraDevice(device_type__value: $device_type) {
    edges {
      node {
        id
        name {
          value
        }
        serial_number {
          value
        }
      }
    }
  }
}
"""

variables = {"device_type": "router"}
result = await client.execute_graphql(
    query=query,
    variables=variables
)

devices = result["InfraDevice"]["edges"]
for device in devices:
    print(device["node"]["name"]["value"])

Accessing Attributes

Basic Attribute Access

Attributes are accessed with .value:

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

# Read attribute values
name = device.name.value
serial = device.serial_number.value
height = device.height.value  # Integer/Number
is_active = device.is_active.value  # Boolean

Attribute Types

Different attribute kinds:

# Text
device.name.value  # str

# Number
device.height.value  # int

# Boolean
device.is_active.value  # bool

# DateTime
device.commissioned_at.value  # datetime object

# JSON
device.metadata.value  # dict or list

# IPAddress
device.ip_address.value  # str (IP address)

Accessing Relationships

Single Cardinality Relationships

Relationships with cardinality “one”:

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

# Access the related object's ID
location_id = device.location.id

# Access the related object's attributes (if included)
location_name = device.location.name.value

Multiple Cardinality Relationships

Relationships with cardinality “many”:

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

# Iterate over related objects
for tag in device.tags:
    print(f"Tag: {tag.name.value}")

# Get list of IDs
tag_ids = device.tags.peer_ids  # List of UUIDs

# Count related objects
tag_count = len(device.tags)

Relationship Metadata

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

# Access relationship properties
print(f"Location ID: {device.location.id}")
print(f"Location updated: {device.location.updated_at}")
print(f"From profile: {device.location.is_from_profile}")

Querying on Branches

Query on a Specific Branch

# Query on main branch (default)
device = await client.get(
    kind="InfraDevice",
    id="device-id"
)

# Query on a different branch
device_on_branch = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch="feature-branch"
)

Compare Across Branches

# Get object on main branch
main_device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch="main"
)

# Get same object on feature branch
feature_device = await client.get(
    kind="InfraDevice",
    id="device-id",
    branch="feature-update"
)

# Compare values
if main_device.name.value != feature_device.name.value:
    print(f"Name changed: {main_device.name.value} -> {feature_device.name.value}")

Handling Missing Data

Check for None Values

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

# Optional attributes may be None
if device.description.value is None:
    print("No description set")
else:
    print(f"Description: {device.description.value}")

Handle Missing Relationships

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

# Check if relationship exists
if device.location:
    print(f"Location: {device.location.name.value}")
else:
    print("No location assigned")

Try-Except for Not Found

from infrahub_sdk.exceptions import NodeNotFoundError

try:
    device = await client.get(
        kind="InfraDevice",
        id="nonexistent-id"
    )
except NodeNotFoundError:
    print("Device not found")

Querying Schema

Fetch Schema Information

# Fetch schema for specific namespaces
await client.schema.fetch(namespaces=["Infra", "Network"])

# Get schema for a specific kind
device_schema = await client.schema.get(
    kind="InfraDevice",
    branch="main"
)

print(f"Schema name: {device_schema.name}")
print(f"Namespace: {device_schema.namespace}")
print(f"Attributes: {[attr.name for attr in device_schema.attributes]}")

Check Schema Sync Status

# Wait for schema to converge
await client.schema.wait_until_converged(branch="main")

# Check if schema is in sync
is_synced = await client.schema.in_sync()
if is_synced:
    print("Schema is synchronized")
else:
    print("Schema update in progress")

Performance Optimization

Batch Queries

Query multiple objects efficiently:

device_ids = ["id-1", "id-2", "id-3"]

# Query all devices
devices = []
for device_id in device_ids:
    device = await client.get(kind="InfraDevice", id=device_id)
    devices.append(device)

Limit Included Relationships

Only include what you need:

# Include only required relationships
device = await client.get(
    kind="InfraDevice",
    id="device-id",
    include=["location"]  # Don't include unnecessary relationships
)

Use Pagination

For large datasets, use pagination (see Pagination guide):

# Instead of loading all objects
# all_devices = await client.all(kind="InfraDevice")  # May be slow

# Use pagination for large datasets
offset = 0
limit = 100

while True:
    batch = await client.execute_graphql(
        query="""query GetDevices($offset: Int!, $limit: Int!) {
          InfraDevice(offset: $offset, limit: $limit) {
            edges { node { id name { value } } }
          }
        }""",
        variables={"offset": offset, "limit": limit}
    )
    
    devices = batch["InfraDevice"]["edges"]
    if not devices:
        break
    
    for device in devices:
        print(device["node"]["name"]["value"])
    
    offset += limit

Common Query Patterns

Get Object with Full Context

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

Search by Attribute Value

all_devices = await client.all(kind="InfraDevice")
target_device = next(
    (d for d in all_devices if d.serial_number.value == "SN12345"),
    None
)

if target_device:
    print(f"Found: {target_device.name.value}")

Count Objects

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

active = sum(1 for d in devices if d.is_active.value)
inactive = total - active

print(f"Total: {total}, Active: {active}, Inactive: {inactive}")

Next Steps

Creating Objects

Learn how to create new infrastructure objects

Updating Objects

Update existing objects and attributes

Relationships

Work with object relationships

Pagination

Handle large datasets with pagination