Querying Data

Query and filter data using the Infrahub Python SDK

The Infrahub SDK provides multiple ways to query data from the API, from simple retrieval to complex filtered queries.

Basic Queries

Get by ID

Retrieve a specific object by its ID:

from infrahub_sdk import InfrahubClient

client = InfrahubClient()

device = await client.get(
    kind="InfraDevice",
    id="17f54fe9-fabc-87e5-d38f-c51c71ed91a6"
)

if device:
    print(f"Device: {device.name.value}")
else:
    print("Device not found")

Get by Attribute

Retrieve an object by filtering on attributes:

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

# Get by multiple attributes
device = await client.get(
    kind="InfraDevice",
    name__value="router-01",
    status__value="active"
)

Get All Objects

Retrieve all objects of a specific kind:

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

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

Filtering

Filter by Attributes

Use the filters() method for complex filtering:

# Single filter
active_devices = await client.filters(
    kind="InfraDevice",
    status__value="active"
)

# Multiple filters (AND logic)
edge_routers = await client.filters(
    kind="InfraDevice",
    status__value="active",
    role__name__value="edge"
)

Filter Syntax

The SDK supports Django-style filter lookups:

Filter	Example	Description
__value	name__value="router-01"	Exact value match
__contains	name__value__contains="router"	Contains substring
__in	status__value__in=["active", "maintenance"]	Value in list
__gt	port_count__value__gt=24	Greater than
__gte	port_count__value__gte=24	Greater than or equal
__lt	port_count__value__lt=48	Less than
__lte	port_count__value__lte=48	Less than or equal

Example:

# Devices with more than 24 ports
devices = await client.filters(
    kind="InfraDevice",
    port_count__value__gte=24
)

# Devices with names containing "router"
devices = await client.filters(
    kind="InfraDevice",
    name__value__contains="router"
)

Relationship Filters

Filter by related objects:

# Devices at a specific site
devices = await client.filters(
    kind="InfraDevice",
    site__name__value="NYC"
)

# Devices with specific role and site
devices = await client.filters(
    kind="InfraDevice",
    role__name__value="edge",
    site__name__value="NYC"
)

Working with Branches

Query in Specific Branch

Query data from a specific branch:

# Get from main branch (default)
device = await client.get(
    kind="InfraDevice",
    id="abc-123"
)

# Get from feature branch
device = await client.get(
    branch="feature-new-devices",
    kind="InfraDevice",
    id="abc-123"
)

Compare Across Branches

Query the same object in different branches:

# Get from main
main_device = await client.get(
    branch="main",
    kind="InfraDevice",
    name__value="router-01"
)

# Get from feature branch
feature_device = await client.get(
    branch="feature-updates",
    kind="InfraDevice",
    name__value="router-01"
)

if main_device and feature_device:
    print(f"Main: {main_device.status.value}")
    print(f"Feature: {feature_device.status.value}")

GraphQL Queries

Custom GraphQL Queries

Build custom GraphQL queries for advanced use cases:

from infrahub_sdk.graphql import Query

# Define a custom query
query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None},
                    "status": {"value": None},
                    "site": {
                        "node": {
                            "name": {"value": None}
                        }
                    }
                }
            }
        }
    }
)

# Execute the query
result = await client.execute_graphql(query=query.render_query())

Query with Filters

Add filters to GraphQL queries:

from infrahub_sdk.graphql import Query

query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "@filters": {
                "status__value": "active"
            },
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None}
                }
            }
        }
    }
)

result = await client.execute_graphql(query=query.render_query())

Nested Relationships

Query nested relationships:

from infrahub_sdk.graphql import Query

query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None},
                    "site": {
                        "node": {
                            "name": {"value": None},
                            "region": {
                                "node": {
                                    "name": {"value": None}
                                }
                            }
                        }
                    },
                    "interfaces": {
                        "edges": {
                            "node": {
                                "name": {"value": None},
                                "status": {"value": None}
                            }
                        }
                    }
                }
            }
        }
    }
)

result = await client.execute_graphql(query=query.render_query())

Accessing Node Attributes

Basic Attributes

Access attribute values:

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

# Get value
name = device.name.value
status = device.status.value

# Get metadata
name_source = device.name.source  # Source object ID
name_owner = device.name.owner    # Owner object ID
name_protected = device.name.is_protected  # Protected flag

Relationship Attributes

Access related objects:

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

# Single relationship (peer)
if device.site.peer:
    site = await device.site.peer.fetch()
    print(f"Site: {site.name.value}")

# Many relationship (peers)
if device.interfaces.peers:
    for interface_peer in device.interfaces.peers:
        interface = await interface_peer.fetch()
        print(f"Interface: {interface.name.value}")

Fetching Related Objects

Fetch related objects lazily:

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

# Fetch single relationship
site = await device.site.fetch()
print(f"Device {device.name.value} is at {site.name.value}")

# Fetch multiple relationships
interfaces = await device.interfaces.fetch()
for interface in interfaces:
    print(f"  Interface: {interface.name.value}")

Pagination

Limit Results

Limit the number of results:

from infrahub_sdk.graphql import Query

query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "@limit": 10,  # Limit to 10 results
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None}
                }
            }
        }
    }
)

result = await client.execute_graphql(query=query.render_query())

Offset Results

Skip results for pagination:

query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "@offset": 20,  # Skip first 20 results
            "@limit": 10,   # Get next 10
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None}
                }
            }
        }
    }
)

result = await client.execute_graphql(query=query.render_query())

Sorting

Order Results

Sort query results:

from infrahub_sdk.graphql import Query
from infrahub_sdk.types import Order

query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "@order_by": {
                "name__value": Order.ASC
            },
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None}
                }
            }
        }
    }
)

result = await client.execute_graphql(query=query.render_query())

Multiple Sort Fields

Sort by multiple fields:

query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "@order_by": {
                "site__name__value": Order.ASC,
                "name__value": Order.DESC
            },
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None}
                }
            }
        }
    }
)

Query Performance

Select Only Required Fields

Reduce payload size by selecting only needed fields:

# Inefficient - fetches all fields
devices = await client.all(kind="InfraDevice")

# Efficient - fetch only needed fields
from infrahub_sdk.graphql import Query

query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None}
                    # Only fetch id and name
                }
            }
        }
    }
)

Batch Fetching

Fetch multiple objects efficiently:

# Get multiple IDs at once
device_ids = ["id-1", "id-2", "id-3"]

query = Query(
    name="InfraDevice",
    query={
        "InfraDevice": {
            "@filters": {
                "ids": device_ids
            },
            "edges": {
                "node": {
                    "id": None,
                    "name": {"value": None}
                }
            }
        }
    }
)

result = await client.execute_graphql(query=query.render_query())

Error Handling

Handle query errors gracefully:

from infrahub_sdk.exceptions import GraphQLError, Error

try:
    device = await client.get(
        kind="InfraDevice",
        name__value="router-01"
    )
    if not device:
        print("Device not found")
except GraphQLError as e:
    print(f"GraphQL error: {e.message}")
    print(f"Query: {e.query}")
except Error as e:
    print(f"SDK error: {e.message}")

Examples

Find Devices at a Site

async def get_site_devices(site_name: str):
    client = InfrahubClient()
    
    devices = await client.filters(
        kind="InfraDevice",
        site__name__value=site_name
    )
    
    print(f"Devices at {site_name}:")
    for device in devices:
        print(f"  - {device.name.value}")

Get Active Interfaces

async def get_active_interfaces():
    client = InfrahubClient()
    
    interfaces = await client.filters(
        kind="InfraInterface",
        status__value="active",
        enabled__value=True
    )
    
    for interface in interfaces:
        device = await interface.device.fetch()
        print(f"{device.name.value} - {interface.name.value}")

Complex Query with Relationships

async def get_edge_routers_with_interfaces():
    client = InfrahubClient()
    
    from infrahub_sdk.graphql import Query
    
    query = Query(
        name="InfraDevice",
        query={
            "InfraDevice": {
                "@filters": {
                    "role__name__value": "edge"
                },
                "edges": {
                    "node": {
                        "id": None,
                        "name": {"value": None},
                        "site": {
                            "node": {
                                "name": {"value": None}
                            }
                        },
                        "interfaces": {
                            "@filters": {
                                "status__value": "active"
                            },
                            "edges": {
                                "node": {
                                    "name": {"value": None},
                                    "speed": {"value": None}
                                }
                            }
                        }
                    }
                }
            }
        }
    )
    
    result = await client.execute_graphql(query=query.render_query())
    return result

Next Steps

Mutations

Learn how to create and update data

Batch Operations

Perform efficient bulk operations