[Phase 2] Create Infrahub Client for Fabric Intent #42

New Issue

Damien · 2026-02-05T15:31:44Z

Damien commented

2026-02-05 15:31:44 +00:00

Description

Create a Python client to interact with Infrahub via GraphQL and SDK, replacing the previous NetBox client.

Context

With Infrahub as Source of Truth, we need a client layer to fetch intent data for the reconciler and orchestration workflows. This client wraps the Infrahub SDK and provides domain-specific methods for fabric operations.

Tasks

Setup infrahub-sdk dependency
Create base client with authentication and connection handling
Implement device intent fetching methods
Implement VLAN/VXLAN intent methods
Implement BGP intent methods
Implement VRF intent methods
Implement MLAG intent methods
Add caching layer for performance
Create unit tests with mocked responses

Client Architecture

from infrahub_sdk import InfrahubClient

class FabricInfrahubClient:
    """Client for fetching fabric intent from Infrahub."""
    
    def __init__(self, url: str, api_token: str, branch: str = "main"):
        self.client = InfrahubClient(address=url, api_token=api_token)
        self.branch = branch
    
    async def get_device(self, name: str) -> dict:
        """Fetch device with all relationships."""
        pass
    
    async def get_device_vlans(self, device: str) -> list[dict]:
        """Fetch VLANs assigned to a device."""
        pass
    
    async def get_device_bgp_sessions(self, device: str) -> list[dict]:
        """Fetch BGP sessions for a device."""
        pass
    
    async def get_device_vrfs(self, device: str) -> list[dict]:
        """Fetch VRFs with L3VNI mappings."""
        pass
    
    async def get_mlag_domain(self, device: str) -> dict | None:
        """Fetch MLAG domain config for a device."""
        pass

Example Usage

async with FabricInfrahubClient(
    url="https://infrahub.example.com",
    api_token=os.environ["INFRAHUB_TOKEN"]
) as client:
    # Fetch all intent for a device
    device = await client.get_device("leaf1")
    vlans = await client.get_device_vlans("leaf1")
    bgp = await client.get_device_bgp_sessions("leaf1")
    
    # Use with transforms
    for vlan in vlans:
        yang_payload = render_vlan_transform(vlan)

Output Files

src/infrahub/
├── __init__.py
├── client.py          # Main client class
├── queries/           # Reusable GraphQL queries
│   ├── device.gql
│   ├── vlan.gql
│   ├── bgp.gql
│   └── vrf.gql
└── models.py          # Pydantic models for responses
tests/
└── test_infrahub_client.py

Acceptance Criteria

Client connects to Infrahub with token auth
All fabric intent types fetchable
Async/await support
Branch selection for proposed changes
Unit tests with mocked SDK responses
Type hints and Pydantic models

Migration Notes

NetBox Client	Infrahub Client
`pynetbox`	`infrahub-sdk`
REST API	GraphQL
`get_device_vlans()`	`get_device_vlans()` (same interface)
Sync	Async native

Depends on: #41 (Schema must exist to query)
Replaces: src/netbox/client.py (removed in e469466)
Used by: Reconciler, Prefect flows

## Description Create a Python client to interact with Infrahub via GraphQL and SDK, replacing the previous NetBox client. ## Context With Infrahub as Source of Truth, we need a client layer to fetch intent data for the reconciler and orchestration workflows. This client wraps the Infrahub SDK and provides domain-specific methods for fabric operations. ## Tasks - [ ] Setup `infrahub-sdk` dependency - [ ] Create base client with authentication and connection handling - [ ] Implement device intent fetching methods - [ ] Implement VLAN/VXLAN intent methods - [ ] Implement BGP intent methods - [ ] Implement VRF intent methods - [ ] Implement MLAG intent methods - [ ] Add caching layer for performance - [ ] Create unit tests with mocked responses ## Client Architecture ```python from infrahub_sdk import InfrahubClient class FabricInfrahubClient: """Client for fetching fabric intent from Infrahub.""" def __init__(self, url: str, api_token: str, branch: str = "main"): self.client = InfrahubClient(address=url, api_token=api_token) self.branch = branch async def get_device(self, name: str) -> dict: """Fetch device with all relationships.""" pass async def get_device_vlans(self, device: str) -> list[dict]: """Fetch VLANs assigned to a device.""" pass async def get_device_bgp_sessions(self, device: str) -> list[dict]: """Fetch BGP sessions for a device.""" pass async def get_device_vrfs(self, device: str) -> list[dict]: """Fetch VRFs with L3VNI mappings.""" pass async def get_mlag_domain(self, device: str) -> dict | None: """Fetch MLAG domain config for a device.""" pass ``` ## Example Usage ```python async with FabricInfrahubClient( url="https://infrahub.example.com", api_token=os.environ["INFRAHUB_TOKEN"] ) as client: # Fetch all intent for a device device = await client.get_device("leaf1") vlans = await client.get_device_vlans("leaf1") bgp = await client.get_device_bgp_sessions("leaf1") # Use with transforms for vlan in vlans: yang_payload = render_vlan_transform(vlan) ``` ## Output Files ``` src/infrahub/ ├── __init__.py ├── client.py # Main client class ├── queries/ # Reusable GraphQL queries │ ├── device.gql │ ├── vlan.gql │ ├── bgp.gql │ └── vrf.gql └── models.py # Pydantic models for responses tests/ └── test_infrahub_client.py ``` ## Acceptance Criteria - [ ] Client connects to Infrahub with token auth - [ ] All fabric intent types fetchable - [ ] Async/await support - [ ] Branch selection for proposed changes - [ ] Unit tests with mocked SDK responses - [ ] Type hints and Pydantic models ## Migration Notes | NetBox Client | Infrahub Client | |---------------|-----------------| | `pynetbox` | `infrahub-sdk` | | REST API | GraphQL | | `get_device_vlans()` | `get_device_vlans()` (same interface) | | Sync | Async native | ## Related - Depends on: #41 (Schema must exist to query) - Replaces: `src/netbox/client.py` (removed in e469466) - Used by: Reconciler, Prefect flows

Damien added the phase-2-minimal-reconciler label 2026-02-05 15:31:50 +00:00

Damien referenced this issue

2026-02-09 18:08:50 +00:00

[Schema] Fix non-unique human_friendly_id on BGPPeerGroup and BGPSession #43

Damien added this to the Fabric Orchestrator project 2026-02-09 19:03:29 +00:00

Damien referenced this issue

2026-02-13 19:26:21 +00:00

[Phase 2] Create Infrahub Object Files for Reference Topology Data #52

Damien moved this to In Progress in Fabric Orchestrator on 2026-02-25 13:59:41 +00:00

Damien referenced this issue from a commit

2026-02-26 13:03:39 +00:00

feat(infrahub): add Infrahub client for fabric intent (#42)