How to Create Your First Automation Workflow

This guide walks you through building a complete network automation workflow using Infrahub as your inventory source and artifact management system. You'll retrieve and deploy device configurations managed by Infrahub.

What you'll accomplish

By the end of this guide, you'll have:

• Connected Nornir to your Infrahub instance
• Retrieved device configurations from Infrahub artifacts
• Used dynamic device groups for targeted operations
• Deployed configurations managed by Infrahub
• Understood how to integrate Nornir with Infrahub's artifact system

Prerequisites

• Infrahub instance with network devices and artifact definitions configured
  • You can use the Infrahub sandbox instance to try this workflow
• Nornir-Infrahub plugin installed (pip install nornir-infrahub)
• Connection plugin for your devices: pip install nornir-netmiko or nornir-napalm
• Basic Python and Nornir knowledge
• Access to test network devices (or simulation)

Step 1: Set up your project structure

Create a project directory with the following structure:

mkdir network-automation
cd network-automation

# Create directory structure
mkdir -p configs
touch config.yaml
touch workflow.py

Step 2: Configure Nornir with Infrahub

Create your config.yaml file:

---
inventory:
  plugin: InfrahubInventory
  options:
    address: "https://sandbox.infrahub.app"   # Infrahub instance URL
    token: "1808d43b-c370-48e8-d0ef-c51781c02ddf"           # Replace with your actual token
    branch: "main"                         # Infrahub branch to use

    # Define which Infrahub node type represents network devices
    # Using the schema example above
    host_node:
      kind: "InfraDevice"                  # Matches the Device node with Infra namespace

    # Map Infrahub attributes to Nornir host properties
    schema_mappings:
      - name: "hostname"
        mapping: "name"                    # Use device name as hostname
      - name: "platform"
        mapping: "platform.nornir_platform" # Platform attribute for Nornir compatibility

    # Create Nornir groups based on Infrahub attributes
    # Note: Disabled due to some devices missing site/role in sandbox
    # group_mappings:
    #   - "site.name"
    #   - "role.name"

runner:
  plugin: threaded
  options:
    num_workers: 20

Step 3: Understand Infrahub artifact system

Before proceeding, ensure your Infrahub instance has artifact definitions configured. In this workflow, we'll use Infrahub's artifact system to retrieve generated configurations rather than creating templates locally.

Infrahub artifacts provide:

• Centralized Templates: Jinja2 templates stored in Git repositories
• Data-Driven Generation: Configurations generated from your Infrahub data model
• Version Control: All templates and generated artifacts are versioned
• Consistency: Ensure all network configurations follow organizational standards

The Nornir-Infrahub plugin provides tasks to interact with these artifacts:

• get_artifact(): Retrieve generated configuration content
• generate_artifacts(): Trigger artifact generation for targets
• regenerate_host_artifact(): Regenerate configuration for a specific device

Step 4: Build the workflow script

Create workflow.py with your automation workflow:

#!/usr/bin/env python3
"""Network automation workflow using Nornir and Infrahub artifacts.

This module implements a complete configuration management workflow that:
1. Connects to Infrahub to load device inventory
2. Regenerates device configurations using Infrahub's artifact system
3. Retrieves and validates the generated configurations
4. Deploys configurations to network devices (with user confirmation)

The workflow is designed to work with the Infrahub sandbox environment and
targets edge devices that have the "Startup Config for Edge devices" artifact
definition configured.

Requirements:
    - nornir: Core automation framework
    - nornir-infrahub: Infrahub inventory and task plugins
    - nornir-netmiko: Device configuration deployment
    - nornir-utils: Result printing utilities

Example:
    Run the workflow from the command line::

        $ python workflow.py

    Or import and customize::

        from workflow import main
        main()

Note:
    The backup functionality requires a "running-config" artifact definition
    in Infrahub, which is not available in the public sandbox.
"""

import logging
from datetime import datetime
from pathlib import Path

from nornir import InitNornir
from nornir.core.task import Task, Result
from nornir_utils.plugins.functions import print_result
from nornir_netmiko.tasks import netmiko_send_config
from nornir_infrahub.plugins.tasks import (
    get_artifact,
    regenerate_host_artifact,
)

# Configure logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


def retrieve_configuration(task: Task) -> Result:
    """Retrieve device configuration from Infrahub artifacts.

    Fetches the generated configuration artifact from Infrahub and saves it
    locally for validation and deployment. The configuration is saved to
    the `configs/` directory with the filename `{hostname}.cfg`.

    Args:
        task: Nornir task object containing host information and Infrahub client.

    Returns:
        Result object with:
            - On success: Path to saved configuration file
            - On failure: Error message describing the issue

    Raises:
        No exceptions are raised; all errors are captured in the Result object.
    """
    try:
        # Get the configuration artifact from Infrahub
        # Using the artifact name from the sandbox
        result = get_artifact(
            task,
            artifact="Startup Config for Edge devices",
        )

        if result.failed:
            return Result(
                host=task.host,
                failed=True,
                result=f"Failed to retrieve artifact: {result.result}",
            )

        # Check if we got actual content
        if not result.result or result.result.strip() == "":
            return Result(
                host=task.host,
                failed=True,
                result="Artifact exists but has no content",
            )

        # Save configuration locally for deployment
        output_dir = Path("configs")
        output_dir.mkdir(exist_ok=True)

        config_file = output_dir / f"{task.host.name}.cfg"
        config_file.write_text(result.result)

        return Result(
            host=task.host,
            result=f"Configuration retrieved from Infrahub and saved to {config_file}",
        )

    except Exception as e:
        error_msg = str(e)
        if "NodeNotFoundError" in error_msg or "Unable to find" in error_msg:
            return Result(
                host=task.host,
                failed=True,
                result="No artifact found for this device (may not be configured in Infrahub)",
            )
        return Result(
            host=task.host,
            failed=True,
            result=f"Failed to retrieve configuration: {error_msg}",
        )


def regenerate_configuration(task: Task) -> Result:
    """Trigger regeneration of device configuration in Infrahub.

    Requests Infrahub to regenerate the configuration artifact for the host.
    This ensures the latest data from Infrahub is used to generate a fresh
    configuration before retrieval.

    Args:
        task: Nornir task object containing host information and Infrahub client.

    Returns:
        Result object with:
            - On success: Confirmation message
            - On failure: Error message describing the issue
    """
    try:
        # Regenerate the artifact in Infrahub
        result = regenerate_host_artifact(
            task,
            artifact="Startup Config for Edge devices",
        )

        if result.failed:
            return Result(
                host=task.host,
                failed=True,
                result=f"Failed to regenerate artifact: {result.result}",
            )

        return Result(
            host=task.host, result="Configuration regenerated successfully in Infrahub"
        )

    except Exception as e:
        error_msg = str(e)
        if "NodeNotFoundError" in error_msg or "Unable to find" in error_msg:
            return Result(
                host=task.host,
                failed=True,
                result="No artifact definition found for this device",
            )
        return Result(
            host=task.host,
            failed=True,
            result=f"Failed to regenerate configuration: {error_msg}",
        )


def validate_configuration(task: Task) -> Result:
    """Validate the retrieved configuration before deployment.

    Performs basic validation checks on the configuration file to ensure
    it is suitable for deployment. Currently validates that the configuration
    is not empty.

    Args:
        task: Nornir task object containing host information.

    Returns:
        Result object with:
            - On success: Validation passed message with file size
            - On failure: List of failed validation checks
    """
    config_file = Path("configs") / f"{task.host.name}.cfg"

    if not config_file.exists():
        return Result(
            host=task.host, failed=True, result="Configuration file not found"
        )

    config_content = config_file.read_text()

    # Basic validation checks - relaxed for sandbox artifacts
    validations = {
        "not_empty": len(config_content.strip()) > 0,
    }

    failed_checks = [check for check, passed in validations.items() if not passed]

    if failed_checks:
        return Result(
            host=task.host,
            failed=True,
            result=f"Validation failed: {', '.join(failed_checks)}",
        )

    return Result(
        host=task.host,
        result=f"Configuration validation passed ({len(config_content)} bytes)",
    )


def deploy_configuration(task: Task) -> Result:
    """Deploy configuration to the network device via Netmiko.

    Reads the configuration from the local `configs/` directory and pushes
    it to the device using Netmiko's send_config method. Requires the device
    to have proper connection parameters configured in the Nornir inventory.

    Args:
        task: Nornir task object containing host information and connection details.

    Returns:
        Result object with:
            - On success: Deployment confirmation message
            - On failure: Error message from Netmiko or file not found error
    """
    config_file = Path("configs") / f"{task.host.name}.cfg"

    if not config_file.exists():
        return Result(
            host=task.host, failed=True, result="Configuration file not found"
        )

    # Read configuration
    config_commands = config_file.read_text().splitlines()

    # Deploy via Netmiko
    netmiko_result = netmiko_send_config(task=task, config_commands=config_commands)

    if netmiko_result.failed:
        return Result(
            host=task.host,
            failed=True,
            result=f"Deployment failed: {netmiko_result.result}",
        )

    return Result(host=task.host, result="Configuration deployed successfully")


def backup_current_config(task: Task) -> Result:
    """Backup current device configuration using Infrahub artifacts.

    Triggers generation of a running-config artifact in Infrahub, then
    retrieves and saves it locally. Backups are stored in timestamped
    directories under `backups/`.

    Note:
        This function requires a "running-config" artifact definition in
        Infrahub. The public sandbox does not have this artifact configured,
        so this step is skipped in the main workflow.

    Args:
        task: Nornir task object containing host information and Infrahub client.

    Returns:
        Result object with:
            - On success: Path to the backup file
            - On failure: Error message describing the issue
    """
    # Trigger artifact generation for this host
    result = regenerate_host_artifact(task=task, artifact="running-config")

    if result.failed:
        return Result(
            host=task.host, failed=True, result="Failed to generate backup artifact"
        )

    # Retrieve the generated artifact
    artifact_result = get_artifact(task=task, artifact="running-config")

    # Save backup locally
    backup_dir = Path("backups") / datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_dir.mkdir(parents=True, exist_ok=True)

    backup_file = backup_dir / f"{task.host.name}_running.cfg"
    backup_file.write_text(artifact_result.result)

    return Result(host=task.host, result=f"Configuration backed up to {backup_file}")


def main() -> int:
    """Execute the complete network automation workflow.

    Orchestrates the following steps:
        1. Load device inventory from Infrahub
        2. Filter to edge devices (which have startup config artifacts)
        3. Skip backup (not available in sandbox)
        4. Regenerate configurations in Infrahub
        5. Retrieve configurations to local `configs/` directory
        6. Validate all retrieved configurations
        7. Deploy configurations (requires user confirmation)

    The workflow will abort if any validation failures occur, preventing
    deployment of invalid configurations.

    Returns:
        Exit code: 0 for success, 1 for validation failures.
    """
    # Initialize Nornir
    nr = InitNornir(config_file="config.yaml")

    logger.info(f"Loaded {len(nr.inventory.hosts)} hosts from Infrahub")

    # Filter to only edge devices (which have the Startup Config artifact)
    edge_devices = nr.filter(filter_func=lambda h: "edge" in h.name)
    logger.info(
        f"Filtered to {len(edge_devices.inventory.hosts)} edge devices with artifacts"
    )

    # Step 1: Skip backup (no running-config artifact in sandbox)
    logger.info("Step 1: Skipping backup (no running-config artifact in sandbox)")

    # Step 2: Regenerate configurations in Infrahub
    logger.info("Step 2: Regenerating configurations in Infrahub...")
    regen_results = edge_devices.run(task=regenerate_configuration)
    print_result(regen_results)

    # Step 3: Retrieve updated configurations from Infrahub
    logger.info("Step 3: Retrieving configurations from Infrahub...")
    retrieve_results = edge_devices.run(task=retrieve_configuration)
    print_result(retrieve_results)

    # Step 4: Validate configurations
    logger.info("Step 4: Validating configurations...")
    validation_results = edge_devices.run(task=validate_configuration)
    print_result(validation_results)

    # Check for validation failures
    failed_hosts = [
        host for host, result in validation_results.items() if result.failed
    ]

    if failed_hosts:
        logger.error(f"Validation failed for hosts: {failed_hosts}")
        logger.info("Aborting deployment due to validation failures")
        return 1

    # Step 5: Deploy configurations (with confirmation)
    logger.info("Step 5: Ready to deploy configurations")
    response = input("Deploy configurations to all devices? (yes/no): ")

    if response.lower() == "yes":
        logger.info("Deploying configurations...")
        deploy_results = edge_devices.run(task=deploy_configuration)
        print_result(deploy_results)

        # Generate completion report
        successful = len([r for r in deploy_results.values() if not r.failed])
        failed = len([r for r in deploy_results.values() if r.failed])

        logger.info(f"Deployment complete: {successful} successful, {failed} failed")
    else:
        logger.info("Deployment cancelled by user")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Step 5: Execute the workflow

Run your automation workflow:

python workflow.py

The workflow will:

1. Connect to Infrahub and load device inventory
2. Backup current configurations locally
3. Regenerate configurations in Infrahub using artifact system
4. Retrieve updated configurations from Infrahub
5. Validate the retrieved configurations
6. Deploy configurations (with user confirmation)

Validation

Check retrieved configurations

Review the configuration files retrieved from Infrahub in the configs/ directory:

ls -la configs/
cat configs/router1.cfg

Verify backups

Confirm backups were created:

ls -la backups/*/

Monitor deployment

Watch the deployment output for any errors or warnings.

Advanced usage

Filtering devices

Target specific devices or groups:

# Filter by site
site_devices = nr.filter(site_name="chicago")
site_results = site_devices.run(task=deploy_configuration)

# Filter by role
core_devices = nr.filter(role_name="core")
core_results = core_devices.run(task=retrieve_configuration)

# Complex filtering
critical_devices = nr.filter(
    filter_func=lambda h: hasattr(h, 'criticality') and h.criticality == "high"
)

Parallel execution control

Adjust parallelism for different tasks:

# Reduce workers for deployment
nr.config.runner.options["num_workers"] = 2
deploy_results = nr.run(task=deploy_configuration)

# Increase for read-only operations
nr.config.runner.options["num_workers"] = 50
backup_results = nr.run(task=backup_current_config)

Integration with CI/CD

Create a CI-friendly version:

# ci_workflow.py
import sys
import os

def ci_workflow():
    # Get parameters from environment
    target_branch = os.getenv("INFRAHUB_BRANCH", "main")
    target_site = os.getenv("TARGET_SITE", None)

    # Initialize with CI parameters
    nr = InitNornir(
        config_file="config.yaml",
        inventory={
            "options": {
                "branch": target_branch
            }
        }
    )

    # Filter if site specified
    if target_site:
        nr = nr.filter(site_name=target_site)

    # Run workflow without prompts
    results = nr.run(task=retrieve_configuration)

    # Exit with proper code
    failed_count = len([r for r in results.values() if r.failed])
    sys.exit(failed_count)

Related resources

• Understanding Nornir-Infrahub Integration
• Configuring Schema Mappings
• Artifact Lifecycle Management
• Nornir Task Documentation