Skip to main content

Transforms

Developer Guide

Documents the transform implementations. To view artifacts as an operator, see Viewing Artifacts.

Overview​

Transforms convert Infrahub data into usable outputs (configs, documentation, computed attributes). They run on-demand when artifacts are accessed.

Transform types​

  1. Python Transforms - Complex logic, external library calls (PyAVD)
  2. Jinja2 Transforms - Template-based text generation

Every transform this repository registers is a Python transform. One of them, containerlab_topology, renders its output through a Jinja2 template it loads itself; the jinja2_transforms: block of .infrahub.yml is unused here.

Transform architecture​

┌──────────────────┐     ┌────────────────────┐     ┌─────────────────┐
│  GraphQL Query   │ ──▶ │  Transform Class   │ ──▶ │  Artifact       │
│  (*.gql)         │     │  (*Transform)      │     │  (config/doc)   │
└──────────────────┘     └────────────────────┘     └─────────────────┘

Python transforms​

ComputedInterfaceDescription​

File: transforms/computed_interface_description.py

Purpose: Generate human-readable interface descriptions

Input: DcimInterface Output: String like "→ remote-device:Ethernet1"

class ComputedInterfaceDescription(InfrahubTransform):
    async def transform(self, data):
        interface = data["DcimInterface"]["edges"][0]["node"]
        link = interface.get("link")
        if not link:
            return ""

        # Find the remote end
        remote = link["interface_a"] if link["interface_b"]["id"] == interface["id"] else link["interface_b"]
        return f"→ {remote['device']['name']['value']}:{remote['name']['value']}"

CablingPlan​

File: transforms/cabling_plan.py

Purpose: Generate CSV cabling documentation for a fabric

Input: NetworkFabric Output: CSV file with all connections

Source Device,Source Interface,Destination Device,Destination Interface,Link Type
spine-A1-1,Ethernet1,super-spine-A-1,Ethernet1,uplink
leaf-A1-01-1,Ethernet49,spine-A1-1,Ethernet1,uplink

AvdEosConfigTransform​

File: transforms/avd_eos_config.py

Purpose: Convert AVD structured config to EOS CLI

Input: DcimDevice (with AvdArtifact) Output: EOS CLI configuration

class AvdEosConfigTransform(InfrahubTransform):
    async def transform(self, data):
        device = data["DcimDevice"]["edges"][0]["node"]
        artifact = device["avd_artifact"]["node"]

        if not artifact["structured_config_identifier"]["value"]:
            return "! No structured config available"

        config = await self.client.object_store.get(
            identifier=artifact["structured_config_identifier"]["value"]
        )
        return pyavd.get_device_config(json.loads(config))

AvdFabricDocTransform​

File: transforms/avd_fabric_doc.py

Purpose: Generate fabric-wide documentation

Input: NetworkFabric Output: Markdown documentation

class AvdFabricDocTransform(InfrahubTransform):
    async def transform(self, data):
        # Collect all device hostvars and structured configs
        all_hostvars = {}
        all_structured = {}

        for device in devices:
            hostvars = await self.client.object_store.get(...)
            structured = await self.client.object_store.get(...)
            all_hostvars[device.name] = hostvars
            all_structured[device.name] = structured

        avd_facts = pyavd.get_avd_facts(all_hostvars)
        return pyavd.get_fabric_documentation(
            avd_facts, all_structured, fabric_name
        )

AvdDeviceDocTransform​

File: transforms/avd_device_doc.py

Purpose: Generate per-device documentation

Input: DcimDevice Output: Markdown documentation for single device

AvdAntaCatalogTransform​

File: transforms/avd_anta_catalog.py

Purpose: Render a per-device ANTA test catalog from the stored structured config

Input: DcimDevice Output: YAML catalog, or a one-line marker comment

Unlike EOS config rendering, catalog generation needs fabric-wide data, so the transform gathers every sibling device's structured config in the same fabric into one AVDFabricData before calling pyavd.get_device_test_catalog().

File metadata is read in one collection per catalog to avoid repeated graph requests exceeding the task timeout. Complete configs are then downloaded and validated sequentially, with fresh metadata and contents on every execution.

It is gated by the fabric's anta_enabled flag. When ANTA is disabled — or the device has no fabric, or no structured config — the transform returns a marker comment instead of a catalog, so the artifact renders successfully and says why it is empty:

# ANTA disabled for fabric Fabric-L3LS-Multi-Domain

ContainerLabTopology​

File: transforms/containerlab_topology.py

Purpose: Render a ContainerLab topology file for a whole fabric

Input: NetworkFabric Output: YAML topology (topology.clab.yml shape)

The transform uses two queries — containerlab_topology for the fabric's devices and containerlab_link_endpoints to resolve link endpoints in batches — and renders through the transforms/templates/containerlab_topology.j2 template. Node kinds, container images, and the interface-mapping bind come from schema attributes (DcimPlatform.containerlab_os, DcimPlatform.containerlab_image, DcimDeviceType.containerlab_interface_mapping), so changing the cEOS version is a data change rather than a code change.

Nodes and links are emitted in a stable sorted order, so two renders of unchanged data are byte-identical. See the ContainerLab page for the full artifact shape, the role-selection rules, and how to deploy the topology.

CVWorkspaceSubmissionWebhookPayload​

File: transforms/cv_workspace_submission_webhook.py

Purpose: Build the JSON body for the CloudVision workspace-submission CoreCustomWebhook

Input: CloudvisionWorkspace Output: JSON object — check name, proposed-change ID, and one entry per linked workspace with its ID, status, URL, and fabric name

This transform has no artifact definition: it is referenced by the webhook rather than rendered to the object store. See Checks for how it fits the CloudVision validation pipeline.

Query classes​

Each transform has Pydantic models for type-safe query parsing. These *_query.py files are generated, not hand-written — regenerate them whenever the .gql query or the schema changes:

uv run infrahubctl graphql generate-return-types transforms/computed_interface_description.gql

This reads schema.graphql at the repo root (refresh with uv run infrahubctl graphql export-schema --destination schema.graphql when needed) and emits the matching *_query.py next to the query file.

Shape of a typical generated class:

# transforms/computed_interface_description_query.py  (generated)

class InterfaceLink(BaseModel):
    interface_a: InterfaceNode
    interface_b: InterfaceNode

class InterfaceNode(BaseModel):
    id: str
    name: ValueWrapper[str]
    device: DeviceRef
    link: InterfaceLink | None

Artifacts​

Artifacts are the output files generated by transforms.

Artifact definitions​

Defined in .infrahub.yml:

artifact_definitions:
  - name: cabling_plan
    targets: fabrics
    transformation: cabling_plan
    content_type: text/csv

  - name: avd_eos_configuration
    targets: avd_devices
    transformation: avd_eos_config
    content_type: text/plain

  - name: avd_fabric_documentation
    targets: fabrics
    transformation: avd_fabric_doc
    content_type: text/markdown

  - name: avd_device_documentation
    targets: avd_devices
    transformation: avd_device_doc
    content_type: text/markdown

  - name: avd_anta_catalog
    targets: avd_devices
    transformation: avd_anta_catalog
    content_type: application/yaml

  - name: containerlab_topology
    targets: fabrics
    transformation: containerlab_topology
    content_type: application/yaml

cv_workspace_submission_webhook_payload is deliberately absent from this block — it renders a webhook body, not an artifact.

Viewing artifacts​

  1. Navigate to target object in Infrahub UI
  2. Click Artifacts tab
  3. Select artifact to view/download

Regenerating artifacts​

Artifacts regenerate automatically when:

  • Underlying data changes
  • Transform code changes
  • Manually triggered via UI

Configuration​

Transforms are registered in .infrahub.yml:

python_transforms:
  - name: computed_interface_description
    class_name: ComputedInterfaceDescription
    file_path: "./transforms/computed_interface_description.py"

  - name: cabling_plan
    class_name: CablingPlan
    file_path: "./transforms/cabling_plan.py"

  - name: avd_eos_config
    class_name: AvdEosConfigTransform
    file_path: "./transforms/avd_eos_config.py"

  - name: avd_fabric_doc
    class_name: AvdFabricDocTransform
    file_path: "./transforms/avd_fabric_doc.py"

  - name: avd_device_doc
    class_name: AvdDeviceDocTransform
    file_path: "./transforms/avd_device_doc.py"

  - name: avd_anta_catalog
    class_name: AvdAntaCatalogTransform
    file_path: "./transforms/avd_anta_catalog.py"

  - name: containerlab_topology
    class_name: ContainerLabTopology
    file_path: "./transforms/containerlab_topology.py"

  - name: cv_workspace_submission_webhook_payload
    class_name: CVWorkspaceSubmissionWebhookPayload
    file_path: "./transforms/cv_workspace_submission_webhook.py"
    convert_query_response: false

File structure​

transforms/
├── computed_interface_description.py      # Interface description
├── computed_interface_description.gql     # Interface query
├── computed_interface_description_query.py # Pydantic models
├── cabling_plan.py                        # Cabling plan CSV
├── fabric_cabling_plan.gql                # Fabric cabling query
├── fabric_cabling_plan_query.py           # Pydantic models
├── avd_eos_config.py                      # EOS config transform
├── avd_device_config.gql                  # Device config query
├── avd_device_config_query.py             # Pydantic models
├── avd_fabric_doc.py                      # Fabric documentation
├── avd_device_doc.py                      # Device documentation
├── avd_fabric_devices.gql                 # Fabric devices query
├── avd_fabric_devices_query.py            # Pydantic models
├── avd_anta_catalog.py                    # ANTA catalog transform
├── avd_anta_catalog.gql                   # ANTA catalog query
├── avd_anta_catalog_query.py              # Pydantic models
├── containerlab_topology.py               # ContainerLab topology transform
├── containerlab_topology.gql              # Fabric device/link query
├── containerlab_topology_query.py         # Pydantic models
├── containerlab_link_endpoints.gql        # Batched link-endpoint query
├── containerlab_link_endpoints_query.py   # Pydantic models
├── cv_workspace_submission_webhook.py     # CloudVision webhook payload
├── cv_workspace_submission_webhook.gql    # Workspace query
├── cv_workspace_submission_webhook_query.py  # Pydantic models
└── templates/
    └── containerlab_topology.j2           # ContainerLab topology template

Creating new transforms​

Python transform​

  1. Create transform class:
# transforms/my_transform.py
from infrahub_sdk.transforms import InfrahubTransform

class MyTransform(InfrahubTransform):
    query = "my_query"

    async def transform(self, data):
        # Process data
        return "output"
  1. Create GraphQL query:
# transforms/my_query.gql
query MyQuery($device_id: String!) {
  DcimDevice(ids: [$device_id]) {
    edges {
      node {
        name { value }
      }
    }
  }
}
  1. Register in .infrahub.yml:
queries:
  - name: my_query
    file_path: "./transforms/my_query.gql"

python_transforms:
  - name: my_transform
    class_name: MyTransform
    file_path: "./transforms/my_transform.py"

artifact_definitions:
  - name: my_artifact
    targets: devices
    transformation: my_transform

Jinja2 transform​

  1. Create template:
{# transforms/templates/my_template.j2 #}
Output for {{ node.name.value }}
  1. Register in .infrahub.yml:
jinja2_transforms:
  - name: my_jinja_transform
    template_path: "./transforms/templates/my_template.j2"
    query: my_query

Source​