Skip to main content

Developer guide

This guide provides a technical deep-dive into how the Infrahub demo works under the hood. Use this when you want to extend functionality, troubleshoot issues, customize the demo, or understand implementation details.

Project architecture​

The demo follows Infrahub's SDK pattern with five core component types working together:

Schemas → Data → Generators → Transforms → Configurations
↓
Checks (Validation)

Component types​

  1. Schemas (schemas/) - Define data models, relationships, and constraints
  2. Generators (generators/) - Create infrastructure topology programmatically
  3. Transforms (transforms/) - Convert Infrahub data to device configurations
  4. Checks (checks/) - Validate configurations and connectivity
  5. Templates (templates/) - Jinja2 templates for device configurations

All components are registered in .infrahub.yml, which acts as the configuration hub.

Project structure​

infrahub-demo-dc/
├── .infrahub.yml # Component registration
├── checks/ # Validation checks
│ ├── spine.py
│ ├── leaf.py
│ ├── edge.py
│ └── loadbalancer.py
├── objects/ # Demo data
│ ├── bootstrap/ # Initial data (19 files: groups, locations, platforms, roles, devices, etc.)
│ ├── cloud_security/ # Cloud security examples (services, devices, gateways)
│ ├── dc/ # Data center design files
│ │ ├── dc-arista-s.yml # DC-3 design data (Arista)
│ │ ├── dc-cisco-s.yml # DC-2 design data (Cisco)
│ │ ├── dc-cisco-s-border-leafs.yml # Cisco DC with border leafs
│ │ ├── dc-juniper-s.yml # DC-5 design data (Juniper)
│ │ └── dc-sonic-border-leafs.yml # DC-4 design data (SONiC with border leafs)
│ ├── events/ # Event action definitions
│ ├── lb/ # Load balancer configurations
│ ├── pop/ # Point of presence design files
│ │ ├── pop-1.yml # POP-1 design data
│ │ └── pop-2.yml # POP-2 design data
│ └── security/ # Security zones, policies, rules (15 files)
├── generators/ # Topology generators
│ ├── generate_dc.py # Data center generator
│ ├── generate_pop.py # POP generator
│ ├── generate_segment.py # Network segment generator
│ ├── common.py # Shared utilities
│ └── schema_protocols.py # Type protocols
├── menus/ # UI menu definitions
│ └── menu-full.yml # Complete menu
├── queries/ # GraphQL queries
│ ├── config/ # Configuration queries
│ ├── topology/ # Topology queries
│ └── validation/ # Validation queries
├── schemas/ # Data model definitions
│ ├── base/ # Core models
│ │ ├── dcim.yml
│ │ ├── ipam.yml
│ │ ├── location.yml
│ │ └── topology.yml
│ └── extensions/ # Extended models
│ ├── console/
│ ├── routing/
│ ├── security/
│ ├── service/
│ └── topology/
├── docs/ # Documentation (Docusaurus)
│ ├── docs/ # Documentation content (.mdx files)
│ ├── static/ # Static assets
│ ├── docusaurus.config.ts # Docusaurus configuration
│ └── package.json # Node.js dependencies
├── scripts/ # Automation scripts
│ ├── bootstrap.py # Complete setup script
│ ├── populate_security_relationships.py # Security data relationships
│ ├── create_proposed_change.py # Create proposed changes
│ └── get_configs.py # Retrieve device configurations
├── service_catalog/ # Streamlit Service Catalog application
│ ├── pages/ # Streamlit pages
│ │ └── 1_Create_DC.py # DC creation UI
│ ├── utils/ # Utility modules
│ │ ├── api.py # Infrahub API client
│ │ ├── config.py # Configuration
│ │ └── ui.py # UI helpers
│ ├── Home.py # Main application page
│ └── Dockerfile # Container definition
├── templates/ # Jinja2 config templates
├── tests/ # Test suite
│ ├── integration/ # Integration tests
│ ├── smoke/ # Smoke tests
│ └── unit/ # Unit tests
├── transforms/ # Config transforms
│ ├── edge.py
│ ├── leaf.py
│ ├── loadbalancer.py
│ └── spine.py
└── tasks.py # Invoke task definitions

Schemas​

Schemas define the data model using YAML. They specify nodes (object types), attributes, relationships, and constraints.

Schema naming conventions​

  • Nodes: PascalCase (for example, DcimDevice)
  • Attributes: snake_case (for example, device_type)
  • Relationships: snake_case (for example, parent_location)
  • Namespaces: PascalCase (for example, Dcim, Ipam, Service)

Schema example​

nodes:
- name: GenericDevice
namespace: Dcim
description: "A network device"
inherit_from:
- DcimDevice
attributes:
- name: hostname
kind: Text
optional: false
unique: true
- name: device_type
kind: Text
optional: true
relationships:
- name: location
peer: LocationBuilding
cardinality: one
optional: false
- name: interfaces
peer: DcimInterface
cardinality: many
kind: Component

Computed attributes​

Infrahub supports computed attributes that automatically generate values based on other attributes using Jinja2 templates. The demo uses this feature in the BGP schema for Autonomous System names.

Example from schemas/extensions/routing/bgp.yml:

attributes:
- name: name
kind: Text
computed_attribute:
kind: Jinja2
jinja2_template: "AS{{asn__value}}"
read_only: true
optional: false
- name: asn
kind: Number
description: "Autonomous System Number"

When you create an Autonomous System with ASN 65000, the name attribute is automatically computed as "AS65000". This ensures consistency and reduces manual data entry errors.

Benefits of computed attributes:

  • Consistency - Standardized naming conventions enforced automatically
  • Reduced errors - No manual entry of derived values
  • Dynamic updates - Values recompute when dependencies change
  • Read-only enforcement - Prevents manual modification of computed values

Schema types​

The demo includes schemas for:

  • DCIM (Data Center Infrastructure Management) - Devices, interfaces, racks
  • IPAM (IP Address Management) - IP addresses, prefixes, VLANs
  • Location - Sites, buildings, rooms
  • Topology - Data centers, POPs, deployments
  • Routing - BGP, OSPF, routing policies
  • Security - Zones, policies, firewall rules
  • Service - Load balancers, segments, services

Loading schemas​

uv run infrahubctl schema load schemas --branch main

Schemas are loaded into Infrahub and become the foundation for all data.

Generators​

Generators create infrastructure topology programmatically from high-level design inputs. They inherit from InfrahubGenerator and implement the generate() method.

Generator pattern​

from infrahub_sdk.generators import InfrahubGenerator
from typing import Any

class DCTopologyGenerator(InfrahubGenerator):
async def generate(self, data: dict[str, Any]) -> None:
"""Generate data center topology based on design data."""
# 1. Query design data
# 2. Create devices
# 3. Create interfaces
# 4. Create IP addresses
# 5. Create routing configurations
pass

DC generator workflow​

The create_dc generator in generators/generate_dc.py:

  1. Queries the topology design - Reads DC-3 parameters like spine count, leaf count, underlay protocol
  2. Creates resource pools - Sets up IP prefix pools and VLAN pools
  3. Creates devices - Generates spine, leaf, and border-leaf switches with correct roles and platforms
  4. Creates interfaces - Adds physical interfaces, loopbacks, and sub-interfaces
  5. Creates connections - Establishes fabric peering between spines and leaves
  6. Configures routing - Sets up BGP or OSPF underlay and BGP EVPN overlay
  7. Assigns IP addresses - Allocates addresses from pools for all interfaces

Generator registration​

Generators are registered in .infrahub.yml:

generator_definitions:
- name: create_dc
file_path: generators/generate_dc.py
targets: topologies_dc
query: topology_dc
class_name: DCTopologyGenerator
parameters:
name: name__value
  • targets - GraphQL query that selects which objects trigger the generator
  • query - GraphQL query providing input data
  • parameters - Parameters passed from triggering object

Running generators​

Generators can be triggered:

  1. Manually via the web UI (Actions → Generator Definitions)
  2. Via API using GraphQL mutations
  3. Automatically via event actions (if configured)

Transforms​

Transforms convert Infrahub data into device configurations. They inherit from InfrahubTransform and use Jinja2 templates.

Transform pattern​

from infrahub_sdk.transforms import InfrahubTransform
from typing import Any

class SpineTransform(InfrahubTransform):
query = "spine_config" # GraphQL query name

async def transform(self, data: Any) -> Any:
"""Transform InfrahubHub data to spine configuration."""
device = data["DcimDevice"]["edges"][0]["node"]

# Process data
context = self.prepare_context(device)

# Render template
return self.render_template(
template="spine.j2",
data=context
)

def prepare_context(self, device: Any) -> dict[str, Any]:
"""Prepare template context from device data."""
return {
"hostname": device["name"]["value"],
"interfaces": self.process_interfaces(device["interfaces"]),
"bgp": self.process_bgp(device),
}

Transform workflow​

  1. Query data - Fetch device and related data via GraphQL
  2. Process data - Transform into template-friendly structure
  3. Render template - Use Jinja2 to generate configuration
  4. Return artifact - Provide configuration as string

Transform registration​

Transforms are registered in .infrahub.yml:

python_transforms:
- name: spine
class_name: Spine
file_path: transforms/spine.py

artifact_definitions:
- name: spine_config
artifact_name: spine
content_type: text/plain
targets: spines # GraphQL query selecting devices
transformation: spine # Transform name
parameters:
device: name__value

Templates​

Jinja2 templates generate device configurations from structured data.

Template example​

hostname {{ hostname }}

{% for interface in interfaces %}
interface {{ interface.name }}
{% if interface.description %}
description {{ interface.description }}
{% endif %}
{% if interface.ip_address %}
ip address {{ interface.ip_address }}
{% endif %}
{% if interface.enabled %}
no shutdown
{% endif %}
{% endfor %}

router bgp {{ bgp.asn }}
{% for neighbor in bgp.neighbors %}
neighbor {{ neighbor.ip }} remote-as {{ neighbor.asn }}
{% endfor %}

Templates use standard Jinja2 syntax with filters and control structures.

Checks​

Checks validate configurations and connectivity. They inherit from InfrahubCheck and implement the check() method.

Check pattern​

from infrahub_sdk.checks import InfrahubCheck
from typing import Any

class CheckSpine(InfrahubCheck):
query = "spine_validation"

async def check(self, data: Any) -> None:
"""Validate spine device configuration."""
device = data["DcimDevice"]["edges"][0]["node"]

# Validation logic
if not self.has_required_interfaces(device):
self.log_error(
"Missing required interfaces",
object_id=device["id"],
object_type="DcimDevice"
)

if not self.has_bgp_config(device):
self.log_warning(
"BGP not configured",
object_id=device["id"]
)

Check registration​

check_definitions:
- name: validate_spine
class_name: CheckSpine
file_path: checks/spine.py
targets: spines
parameters:
device: name__value

GraphQL queries​

Queries are defined in .gql files and referenced by name in transforms and checks.

Query example​

query GetSpineConfig($device_name: String!) {
DcimDevice(name__value: $device_name) {
edges {
node {
id
name { value }
role { value }
platform { value }
interfaces {
edges {
node {
name { value }
description { value }
ip_addresses {
edges {
node {
address { value }
}
}
}
}
}
}
}
}
}
}

Query registration​

queries:
- name: spine_config
file_path: queries/config/spine.gql

Bootstrap data​

Bootstrap data provides initial objects like locations, platforms, and device types.

Bootstrap structure​

objects/bootstrap/
├── 01_organizations.yml # Organizations
├── 02_asn_pools.yml # BGP ASN pools
├── 03_locations.yml # Sites and buildings
├── 04_platforms.yml # Device platforms
├── 05_roles.yml # Device roles
├── 06_device_types.yml # Device models
├── 07_device_templates.yml # Interface templates
└── ...

Files are numbered to ensure correct loading order due to dependencies.

Interface range expansion​

The bootstrap data uses Infrahub's interface range expansion feature to efficiently define multiple interfaces with compact syntax. This feature automatically expands range notation into individual interfaces.

Example from objects/bootstrap/10_physical_device_templates.yml:

interfaces:
kind: TemplateInterfacePhysical
data:
- template_name: N9K-C9336C-FX2_SPINE_Ethernet1/[1-30]
name: Ethernet1/[1-30]
role: leaf
- template_name: N9K-C9336C-FX2_SPINE_Ethernet1/[31-36]
name: Ethernet1/[31-36]
role: uplink

When loaded, Ethernet1/[1-30] expands to 30 individual interfaces: Ethernet1/1, Ethernet1/2, ... Ethernet1/30. This dramatically reduces YAML verbosity when defining device templates with many interfaces.

Benefits of range expansion:

  • Compact notation - Define dozens of interfaces in a single line
  • Reduced errors - Less repetitive typing means fewer mistakes
  • Simplified maintenance - Update interface ranges without editing individual entries
  • Vendor compatibility - Supports common interface naming patterns (Ethernet, GigabitEthernet, et-, ge-, etc.)

This feature is used extensively throughout the bootstrap data for device templates, physical devices, and topology definitions.

Loading bootstrap data​

uv run infrahubctl object load objects/bootstrap --branch main

Testing​

The suite has three tiers. Only the integration tier needs Docker.

# Unit and specification tests: seconds, no containers
uv run invoke test-unit

# The integration tier every pull request runs
uv run invoke test-integration --tier=core

# Everything, including the extended workflows
uv run invoke test-integration --tier=full

Unit and specification tests​

tests/unit/ covers individual functions with mocked clients, and tests/smoke/ holds SDK plugin specifications: test_graphql.yml runs every GraphQL query registered in .infrahub.yml through the SDK's query checker, and test_j2_transforms.yml renders Jinja templates against recorded input. Neither tier starts a container, so both are fast enough to run on every save.

Integration tests​

The integration suite drives complete workflows against a real Infrahub deployment started by infrahub-testcontainers. It does not use invoke start: it brings up its own throwaway stack, bootstraps it from this repository, and tears it down afterward.

One deployment, shared by every module​

tests/integration/conftest.py declares the deployment at session scope, so the whole suite starts one stack and bootstraps it once. Loading the schema and the bootstrap data takes several minutes, so a stack per module would put most of this coverage out of reach of any reasonable CI timeout.

The consequence is that modules are not independent. They run in file-name order, and later modules build on what earlier ones merged into main:

ModuleWorkflow under test
test_00_bootstrapSchema, menu, objects, repository registration, event actions
test_10_dc_workflowArista DC design, generator, proposed change, merge to main
test_20_artifactsDevice configurations, Containerlab topology, cabling matrix, rack elevation
test_30_graphqlEvery query registered in .infrahub.yml, executed against the live schema
test_40_dc_vendorsA second data center from a different vendor design, with border leafs
test_50_popPOP topology and its virtual devices
test_60_segmentNetwork segment service over the merged fabric
test_70_day2Editing a device, then scaling a deployment out to a larger design
test_80_proposed_changeConflict detection and closing a proposed change without merging

Cross-module dependencies are declared with pytest-dependency, so a module whose prerequisite failed is skipped rather than reported as a second failure.

Tiers and when they run​

Modules are marked core or extended:

  • core runs on every pull request. It must pass on its own, so it never depends on anything an extended module produces.
  • extended runs on a pull request whose branch bumps the Infrahub version (update-infrahub-*, but not update-infrahub-sdk-*), and on a manual run of the CI workflow with its tier input set to full. .github/workflows/ci.yml selects the tier.

Dispatch the tier for a change to a generator or a transform, so those workflows are exercised before it merges. It does not block a merge, because an upstream Infrahub fault currently stops it passing reliably: Infrahub intermittently fails to resolve a member of its own internal generator group, which fails whichever generator is running at the time. Read a failure in that tier against this known fault before concluding the demo is broken.

Testing a specific Infrahub version​

By default the suite tests the Infrahub image matching the installed infrahub-testcontainers version, which is what makes a dependency-bump pull request meaningful. Override it to test any other version:

INFRAHUB_TESTING_IMAGE_VER=1.4.0 uv run invoke test-integration --tier=core

Writing an integration test​

Use the helpers in tests/integration/helpers.py rather than raw GraphQL; they already carry the polling and the failure messages.

from . import constants as c
from . import helpers as h

pytestmark = pytest.mark.extended


@pytest.mark.dependency(name="my_thing_loaded")
async def test_01_load(async_client_main, infrahub_address, infrahub_bootstrap) -> None:
"""One sentence on the property under test."""
await h.ensure_branch(async_client_main, "my-branch")
h.load_objects("objects/dc/dc-juniper-s.yml", address=infrahub_address, branch="my-branch")

Four conventions matter more here than in the unit suite:

  1. Derive expectations from the data model, not from literals. h.expected_role_counts reads a design and returns the device counts it calls for. A hard-coded assert len(devices) == 12 breaks whenever the design changes, and asserts nothing about the generator.
  2. Never treat "the first result appeared" as "the work finished". Every asynchronous step here produces results progressively: a topology generator creates devices in its third of six phases, artifact generation produces one artifact per target-group member, and a proposed-change review opens more than twenty validators. Waiting for "everything that exists is done" is satisfied immediately in all three cases. Use h.wait_for_quiescence, pass an expected count to h.wait_for_artifacts, and let h.wait_for_validations require the validator count to settle.
  3. Say what failed in the assertion message. These tests fail on a machine nobody can log in to. h.merge_proposed_change shows the shape: it quotes the merge task's own logs, the validators that did not conclude successfully, and what each of them objected to.
  4. Assert completion and conclusion separately. A validator that never finishes is an Infrahub problem; a validator that finishes and reports a failure is a demo-data problem.

Diagnosing a failure​

Container logs for infrahub-server and task-worker are attached to any run that had a failure, so the CI log already contains them. Locally, add --log-cli-level=INFO to watch each step as it happens, and -k to narrow the run:

uv run pytest tests/integration/test_10_dc_workflow.py --log-cli-level=INFO

Because the suite shares one deployment, running a single module still bootstraps the stack, and a module that depends on a merged fabric is skipped when run on its own.

Two failure messages are worth recognizing:

  • Unable to find the class CheckEdge from a check or transform does not mean the class is missing. Infrahub reports any exception raised inside a check or transform under that message; the real cause is in the traceback above it, in the container log.
  • Before concluding that a failure is a defect in the demo or in Infrahub, check how the deployment was resourced. A stack trimmed down to share a busy machine produces database timeouts that look like workflow failures.

Code quality​

# Type checking
uv run mypy .

# Linting
uv run ruff check .

# Formatting
uv run ruff format .

# All checks: rumdl, yamllint, ruff, mypy
uv run invoke lint

Development workflow​

Setting up for development​

# Clone repository
git clone https://github.com/opsmill/infrahub-demo-dc.git
cd infrahub-demo-dc

# Install dependencies
uv sync

# Start Infrahub
uv run invoke start

# Optional: Enable Service Catalog in .env
echo "INFRAHUB_SERVICE_CATALOG=true" >> .env
uv run invoke restart-containers

# Load bootstrap data
uv run invoke bootstrap

Making changes​

  1. Create a feature branch in Git
  2. Modify code (generators, transforms, checks, schemas, Service Catalog)
  3. Add tests for new functionality
  4. Run quality checks (uv run invoke lint)
  5. Test locally in Infrahub
    • For Service Catalog changes: use uv run invoke start --rebuild
  6. Commit changes with descriptive messages
  7. Create pull request for review

Adding a new generator​

  1. Create Python file in generators/
  2. Implement InfrahubGenerator class
  3. Register in .infrahub.yml under generator_definitions
  4. Declare any helper module it imports under watch.files, so that editing the helper re-runs the generator. Infrahub reads the entry point, so a module imported at runtime is invisible to it
  5. Create associated GraphQL query in queries/
  6. Add unit tests
  7. Test manually in Infrahub

Adding a new transform​

  1. Create Python file in transforms/
  2. Implement InfrahubTransform class
  3. Create Jinja2 template in templates/
  4. Register in .infrahub.yml under python_transforms and artifact_definitions
  5. Declare under watch.files any helper module the transform imports, and the template directory when the template is chosen from device data rather than named in the source. Both are invisible to Infrahub, and without them a template edit leaves every rendered artifact stale
  6. Create GraphQL query in queries/
  7. Add unit tests
  8. Test artifact generation

Adding a new check​

  1. Create Python file in checks/
  2. Implement InfrahubCheck class
  3. Register in .infrahub.yml under check_definitions
  4. Create GraphQL query in queries/
  5. Add unit tests
  6. Test in proposed change workflow

Service catalog development​

The Service Catalog is a Streamlit application that runs in a Docker container. When making changes to the Service Catalog code, you need to rebuild the container image.

Making changes to the service catalog​

  1. Edit Service Catalog code in service_catalog/:
    • Home.py - Main landing page
    • pages/1_Create_DC.py - DC creation form
    • utils/ - Utility modules (api.py, config.py, ui.py)
  2. Rebuild and restart the Service Catalog container:
uv run invoke start --rebuild

The --rebuild flag forces Docker to rebuild the Service Catalog image with your code changes before starting the containers.

When to use --rebuild​

Use the --rebuild flag when you modify:

  • Streamlit page files (Home.py, pages/*.py)
  • Service Catalog utilities (service_catalog/utils/)
  • Service Catalog dependencies (if you modify service_catalog/requirements.txt)
  • Service Catalog Dockerfile

Testing service catalog changes​

  1. Make your code changes in service_catalog/
  2. Rebuild and start with uv run invoke start --rebuild
  3. Access the Service Catalog at http://localhost:8501
  4. Test your changes in the web interface
  5. Check logs for errors:
docker logs infrahub-demo-dc-service-catalog-1

Service catalog environment variables​

Configure the Service Catalog behavior via .env:

INFRAHUB_SERVICE_CATALOG=true # Enable the service catalog
DEFAULT_BRANCH=main # Default branch to show
GENERATOR_WAIT_TIME=60 # Seconds to wait for generator
API_TIMEOUT=30 # API request timeout
API_RETRY_COUNT=3 # Number of API retries

Changes to environment variables do not require --rebuild, just restart:

uv run invoke restart-containers

Extending schemas​

Adding new attributes​

nodes:
- name: GenericDevice
namespace: Dcim
attributes:
- name: serial_number # New attribute
kind: Text
optional: true
unique: true

Adding new relationships​

relationships:
- name: backup_device # New relationship
peer: DcimDevice
cardinality: one
optional: true
description: "Backup device for redundancy"

Creating new node types​

nodes:
- name: Router # New node type
namespace: Dcim
inherit_from:
- DcimDevice
attributes:
- name: routing_instance
kind: Text
optional: false

After modifying schemas, reload them:

uv run infrahubctl schema load schemas --branch main

Common development tasks​

Debugging generators​

Add logging to see execution flow:

import logging

logger = logging.getLogger(__name__)

class MyGenerator(InfrahubGenerator):
async def generate(self, data: dict) -> None:
logger.info(f"Processing topology: {data}")
# ... generator logic

Testing transforms locally​

# Create test data
test_data = {
"DcimDevice": {
"edges": [{"node": {"name": {"value": "spine1"}}}]
}
}

# Initialize transform
transform = SpineTransform(client=client)

# Run transform
result = await transform.transform(test_data)
print(result)

Validating templates​

Use Jinja2 directly to test templates:

from jinja2 import Template

template = Template(open("templates/spine.j2").read())
config = template.render(hostname="spine1", interfaces=[...])
print(config)

Additional resources​