NetBox in Python

Query and update NetBox records from Python services, and stay in sync with its webhooks.

NetBox is the source of truth for network infrastructure - devices, racks, IP addresses, VLANs - and it exposes everything through a REST API. Your services query and update it through outgoing REST connections.

Connections

NetBox authenticates with Authorization: Token <key>, which an API key security definition expresses with Authorization as the header name and the Token prefix included in the secret itself. Each NetBox endpoint gets its own outgoing connection, named after what it serves. In enmasse YAML:

security:
  - name: NetBox API Token
    type: apikey
    header: Authorization
    password: Zato_Enmasse_Env.NetBox_API_Token

outgoing_rest:
  - name: NetBox Devices
    host: https://netbox.example.com
    url_path: /api/dcim/devices/
    security: NetBox API Token
    data_format: json

  - name: NetBox IP Addresses
    host: https://netbox.example.com
    url_path: /api/ipam/ip-addresses/
    security: NetBox API Token
    data_format: json

.. with the environment variable holding the prefix and the key together:

export NetBox_API_Token='Token 0123456789abcdef0123456789abcdef01234567'

Query devices

# -*- coding: utf-8 -*-

from zato.server.service import Service

class GetActiveDevices(Service):
    name = 'network.device.get-active'

    def handle(self):

        conn = self.rest['NetBox Devices']

        # The params dict becomes the query string
        response = conn.get(self.cid, params={
            'status': 'active',
            'site': 'dc-warsaw',
        })

        for device in response.data['results']:
            name = device['name']
            primary_ip = device['primary_ip']
            address = primary_ip['address']
            self.logger.info(f'{name} - {address}')

NetBox paginates its responses - response.data['next'] holds the URL of the next page when there is one, and the limit query parameter controls the page size.

Create records

# -*- coding: utf-8 -*-

from zato.server.service import Service

class AllocateIPAddress(Service):
    name = 'network.ipam.allocate-address'

    def handle(self):

        conn = self.rest['NetBox IP Addresses']

        request = {
            'address': '10.20.30.40/24',
            'status': 'active',
            'description': 'Allocated by the provisioning workflow',
        }

        response = conn.post(self.cid, request)

        self.response.payload = {'id': response.data['id']}

Keep NetBox in sync

The typical integration is bidirectional - NetBox stays the source of truth while other systems learn about changes. NetBox's webhooks can call a REST channel whenever a record changes, and your service fans the update out from there:

# -*- coding: utf-8 -*-

from zato.server.service import Service

class OnDeviceChanged(Service):
    name = 'network.device.on-changed'

    def handle(self):

        # The webhook's payload, already deserialized
        event = self.request.payload

        device = event['data']
        name = device['name']
        status = device['status']
        status_value = status['value']

        self.logger.info(f'Device {name} is now {status_value}')

        # Notify downstream systems - monitoring, CMDB, IPAM reconciliation
        self.invoke('network.monitoring.update-target', device=device)

For scheduled reconciliation instead of webhooks, the scheduler invokes a service that reads NetBox and compares it with reality.

See also

FeatureWhat it does
REST outgoing connectionsPools, timeouts and retries for the NetBox API
Receiving webhooksSignature verification and async processing for NetBox events
SchedulerReconcile NetBox with reality on an interval