Netbox Device Import Journey

I spun up a VM back in 2023 and installed Netbox as alternative to the solution I had previously used which was OpenDCIM. Fast forward to 2025 and I’ve only just worked it out how to import the community device repository instead of creating the devices manually.
Here is the script that I used:

#!/usr/bin/env python3
"""
NetBox Device Type Importer
Imports device types from the community devicetype-library
"""

import os
import sys
import yaml
import requests
import argparse
from pathlib import Path

class NetBoxImporter:
    def __init__(self, url, token, verify_ssl=True):
        self.url = url.rstrip('/')
        self.token = token
        self.headers = {
            'Authorization': f'Token {token}',
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }
        self.session = requests.Session()
        self.session.headers.update(self.headers)
        self.session.verify = verify_ssl
        
        # Disable SSL warnings if verification is disabled
        if not verify_ssl:
            import urllib3
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        
    def test_connection(self):
        """Test connection to NetBox API"""
        try:
            response = self.session.get(f'{self.url}/api/')
            response.raise_for_status()
            print("✓ Successfully connected to NetBox API")
            return True
        except requests.exceptions.RequestException as e:
            print(f"✗ Failed to connect to NetBox API: {e}")
            return False
    
    def get_or_create_manufacturer(self, name):
        """Get existing manufacturer or create new one"""
        slug = name.lower().replace(' ', '-').replace('_', '-')
        
        try:
            # Check if manufacturer exists
            print(f"  Checking for existing manufacturer: {name} (slug: {slug})")
            response = self.session.get(f'{self.url}/api/dcim/manufacturers/', 
                                      params={'slug': slug})
            
            print(f"  API response status: {response.status_code}")
            print(f"  Response content length: {len(response.text)}")
            
            if response.status_code != 200:
                print(f"  ✗ API error checking manufacturer: {response.status_code} - {response.text}")
                return None
            
            if not response.text.strip():
                print(f"  ✗ Empty response from NetBox API")
                return None
            
            try:
                response_data = response.json()
            except ValueError as e:
                print(f"  ✗ Failed to parse JSON response: {e}")
                print(f"  Response text (first 200 chars): {response.text[:200]}")
                return None
                
            if response_data.get('count', 0) > 0:
                manufacturer = response_data['results'][0]
                print(f"  Found existing manufacturer: {name}")
                return manufacturer
            
            # Create new manufacturer
            print(f"  Creating new manufacturer: {name}")
            data = {
                'name': name,
                'slug': slug
            }
            
            response = self.session.post(f'{self.url}/api/dcim/manufacturers/', json=data)
            print(f"  Create response status: {response.status_code}")
            print(f"  Create response length: {len(response.text)}")
            
            if response.status_code == 201:
                if not response.text.strip():
                    print(f"  ✗ Empty response when creating manufacturer")
                    return None
                try:
                    manufacturer = response.json()
                    print(f"  ✓ Created manufacturer: {name}")
                    return manufacturer
                except ValueError as e:
                    print(f"  ✗ Failed to parse create response JSON: {e}")
                    return None
            else:
                print(f"  ✗ Failed to create manufacturer {name}: {response.status_code} - {response.text}")
                return None
                
        except Exception as e:
            print(f"  ✗ Exception in get_or_create_manufacturer for {name}: {e}")
            return None
    
    def import_device_type(self, file_path, manufacturer):
        """Import a single device type from YAML file"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read().strip()
                if not content:
                    print(f"  - Skipping empty file: {file_path.name}")
                    return False
                    
            try:
                device_data = yaml.safe_load(content)
            except yaml.YAMLError as e:
                print(f"  ✗ YAML parsing error in {file_path.name}: {e}")
                return False
                
            if not device_data:
                print(f"  - Skipping file with no data: {file_path.name}")
                return False
            
            # Check if device type already exists
            slug = device_data.get('slug')
            if not slug:
                print(f"  ✗ No slug found in {file_path.name}")
                return False
                
            response = self.session.get(f'{self.url}/api/dcim/device-types/',
                                      params={'slug': slug})
            
            if response.status_code == 200 and response.json()['count'] > 0:
                print(f"  - Device type {device_data.get('model', slug)} already exists")
                return True
            
            # Prepare device type data
            device_type_data = {
                'manufacturer': manufacturer['id'],
                'model': device_data.get('model', ''),
                'slug': slug,
                'part_number': device_data.get('part_number', ''),
                'u_height': device_data.get('u_height', 1),
                'is_full_depth': device_data.get('is_full_depth', True),
                'subdevice_role': device_data.get('subdevice_role'),
                'comments': device_data.get('comments', ''),
            }
            
            # Remove None values
            device_type_data = {k: v for k, v in device_type_data.items() if v is not None}
            
            # Create device type
            response = self.session.post(f'{self.url}/api/dcim/device-types/', 
                                       json=device_type_data)
            
            if response.status_code == 201:
                device_type = response.json()
                print(f"  ✓ Created device type: {device_data.get('model', slug)}")
                
                # Import interfaces if they exist
                self.import_interfaces(device_type, device_data.get('interfaces', []))
                
                # Import power ports if they exist  
                self.import_power_ports(device_type, device_data.get('power-ports', []))
                
                # Import console ports if they exist
                self.import_console_ports(device_type, device_data.get('console-ports', []))
                
                return True
            else:
                print(f"  ✗ Failed to create device type {device_data.get('model', slug)}: {response.text}")
                return False
                
        except yaml.YAMLError as e:
            print(f"  ✗ YAML parsing error in {file_path.name}: {e}")
            return False
        except FileNotFoundError:
            print(f"  ✗ File not found: {file_path}")
            return False
        except UnicodeDecodeError as e:
            print(f"  ✗ Encoding error in {file_path.name}: {e}")
            return False
        except Exception as e:
            print(f"  ✗ Unexpected error importing {file_path.name}: {e}")
            return False
    
    def import_interfaces(self, device_type, interfaces):
        """Import interface templates"""
        for interface_data in interfaces:
            data = {
                'device_type': device_type['id'],
                'name': interface_data.get('name', ''),
                'type': interface_data.get('type', 'other'),
                'mgmt_only': interface_data.get('mgmt_only', False),
            }
            
            response = self.session.post(f'{self.url}/api/dcim/interface-templates/', 
                                       json=data)
            if response.status_code != 201:
                print(f"    Warning: Failed to create interface {interface_data.get('name')}")
    
    def import_power_ports(self, device_type, power_ports):
        """Import power port templates"""
        for port_data in power_ports:
            data = {
                'device_type': device_type['id'],
                'name': port_data.get('name', ''),
                'type': port_data.get('type', ''),
                'maximum_draw': port_data.get('maximum_draw'),
                'allocated_draw': port_data.get('allocated_draw'),
            }
            
            # Remove None values
            data = {k: v for k, v in data.items() if v is not None}
            
            response = self.session.post(f'{self.url}/api/dcim/power-port-templates/', 
                                       json=data)
            if response.status_code != 201:
                print(f"    Warning: Failed to create power port {port_data.get('name')}")
    
    def import_console_ports(self, device_type, console_ports):
        """Import console port templates"""
        for port_data in console_ports:
            data = {
                'device_type': device_type['id'],
                'name': port_data.get('name', ''),
                'type': port_data.get('type', ''),
            }
            
            response = self.session.post(f'{self.url}/api/dcim/console-port-templates/', 
                                       json=data)
            if response.status_code != 201:
                print(f"    Warning: Failed to create console port {port_data.get('name')}")
    
    def import_manufacturer_devices(self, library_path, manufacturer_name, debug=False):
        """Import all device types for a specific manufacturer"""
        manufacturer_path = Path(library_path) / manufacturer_name
        
        if not manufacturer_path.exists():
            print(f"Manufacturer directory not found: {manufacturer_path}")
            return
        
        print(f"\nProcessing manufacturer: {manufacturer_name}")
        
        # Get or create manufacturer
        manufacturer = self.get_or_create_manufacturer(manufacturer_name)
        if not manufacturer:
            return
        
        # Import device types
        yaml_files = list(manufacturer_path.glob('*.yaml')) + list(manufacturer_path.glob('*.yml'))
        
        if not yaml_files:
            print(f"  No YAML files found in {manufacturer_path}")
            return
        
        success_count = 0
        error_count = 0
        skip_count = 0
        
        print(f"  Found {len(yaml_files)} YAML files to process")
        
        for yaml_file in yaml_files:
            if debug:
                print(f"  Processing: {yaml_file.name}")
            try:
                result = self.import_device_type(yaml_file, manufacturer)
                if result:
                    success_count += 1
                else:
                    skip_count += 1
            except yaml.YAMLError as e:
                print(f"  ✗ YAML parsing error in {yaml_file.name}: {e}")
                error_count += 1
            except Exception as e:
                print(f"  ✗ Error importing {yaml_file.name}: {e}")
                error_count += 1
        
        print(f"  Results: {success_count} imported, {skip_count} skipped, {error_count} errors out of {len(yaml_files)} files")

def main():
    parser = argparse.ArgumentParser(description='Import NetBox device types from community library')
    parser.add_argument('--url', required=True, help='NetBox URL')
    parser.add_argument('--token', required=True, help='NetBox API token')
    parser.add_argument('--library', required=True, help='Path to devicetype-library/device-types directory')
    parser.add_argument('--manufacturers', nargs='*', help='Specific manufacturers to import (optional)')
    parser.add_argument('--debug', action='store_true', help='Enable debug output')
    parser.add_argument('--no-ssl-verify', action='store_true', help='Disable SSL certificate verification')
    
    args = parser.parse_args()
    
    # Initialize importer
    importer = NetBoxImporter(args.url, args.token, verify_ssl=not args.no_ssl_verify)
    
    # Test connection
    if not importer.test_connection():
        sys.exit(1)
    
    library_path = Path(args.library)
    if not library_path.exists():
        print(f"Library path not found: {library_path}")
        sys.exit(1)
    
    # Get list of manufacturers to import
    if args.manufacturers:
        manufacturers = args.manufacturers
    else:
        manufacturers = [d.name for d in library_path.iterdir() if d.is_dir()]
    
    print(f"Found {len(manufacturers)} manufacturers to import")
    
    # Import each manufacturer
    total_imported = 0
    total_errors = 0
    total_skipped = 0
    
    for manufacturer in manufacturers:
        try:
            result = importer.import_manufacturer_devices(library_path, manufacturer, args.debug)
        except KeyboardInterrupt:
            print("\nImport interrupted by user")
            break
        except Exception as e:
            print(f"Error processing manufacturer {manufacturer}: {e}")
            total_errors += 1
            continue
    
    print(f"\nImport completed!")
    print(f"Summary: Check individual manufacturer results above for detailed statistics")

if __name__ == '__main__':
    main()
sudo python3 import_device_types.py --url YOUR_NETBOX_URL --token YOUR_NETBOX_TOKEN --library /opt/netbox/devicetype-library/device-types --debug --no-ssl-verify

Leave a Reply

Your email address will not be published. Required fields are marked *

This site uses Akismet to reduce spam. Learn how your comment data is processed.