Skip to content

Nautobot NetBox Importer API Package

nautobot_netbox_importer.diffsync

DiffSync adapter and model implementation for nautobot-netbox-importer.

adapters

Adapter classes for loading DiffSyncModels with data from NetBox or Nautobot.

NautobotAdapter

Bases: NautobotAdapter

DiffSync adapter for Nautobot.

Source code in nautobot_netbox_importer/diffsync/adapters/nautobot.py
class NautobotAdapter(_NautobotAdapter):
    """DiffSync adapter for Nautobot."""

    def __init__(self, *args, job=None, sync=None, **kwargs):
        """Initialize Nautobot.

        Args:
            job (object, optional): Nautobot job. Defaults to None.
            sync (object, optional): Nautobot DiffSync. Defaults to None.
        """
        super().__init__(*args, **kwargs)
        self.job = job
        self.sync = sync
__init__(*args, job=None, sync=None, **kwargs)

Initialize Nautobot.

Parameters:

Name Type Description Default
job object

Nautobot job. Defaults to None.

None
sync object

Nautobot DiffSync. Defaults to None.

None
Source code in nautobot_netbox_importer/diffsync/adapters/nautobot.py
def __init__(self, *args, job=None, sync=None, **kwargs):
    """Initialize Nautobot.

    Args:
        job (object, optional): Nautobot job. Defaults to None.
        sync (object, optional): Nautobot DiffSync. Defaults to None.
    """
    super().__init__(*args, **kwargs)
    self.job = job
    self.sync = sync

NetBoxAdapter

Bases: SourceAdapter

NetBox Source Adapter.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
class NetBoxAdapter(SourceAdapter):
    """NetBox Source Adapter."""

    # pylint: disable=keyword-arg-before-vararg
    def __init__(self, input_ref: _FileRef, options: NetBoxImporterOptions, job=None, sync=None, *args, **kwargs):
        """Initialize NetBox Source Adapter."""
        super().__init__(name="NetBox", get_source_data=_get_reader(input_ref), *args, **kwargs)
        self.job = job
        self.sync = sync

        self.options = options
        self.diff_summary: DiffSummary = {}

        for name in GENERATOR_SETUP_MODULES:
            setup = __import__(name, fromlist=["setup"]).setup
            setup(self)

    def load(self) -> None:
        """Load data from NetBox."""
        self.import_data()
        if self.options.fix_powerfeed_locations:
            fix_power_feed_locations(self)
        self.post_import()

    def import_to_nautobot(self) -> None:
        """Import a NetBox export file into Nautobot."""
        commited = False
        try:
            self._atomic_import()
            commited = True
        except _DryRunException:
            logger.warning("Dry-run mode, no data has been imported.")
        except _ValidationIssuesDetected:
            logger.warning("Data validation issues detected, no data has been imported.")

        if commited and self.options.update_paths:
            logger.info("Updating paths ...")
            call_command("trace_paths", no_input=True)
            logger.info(" ... Updating paths completed.")

        if self.options.summary:
            print_summary(self, self.diff_summary, self.options.field_mapping)

    @atomic
    def _atomic_import(self) -> None:
        self.load()

        diff = self.nautobot.sync_from(self)
        self.diff_summary = diff.summary()

        if self.nautobot.validation_issues and not self.options.bypass_data_validation:
            raise _ValidationIssuesDetected("Data validation issues detected, aborting the transaction.")

        if self.options.dry_run:
            raise _DryRunException("Aborting the transaction due to the dry-run mode.")
__init__(input_ref, options, job=None, sync=None, *args, **kwargs)

Initialize NetBox Source Adapter.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
def __init__(self, input_ref: _FileRef, options: NetBoxImporterOptions, job=None, sync=None, *args, **kwargs):
    """Initialize NetBox Source Adapter."""
    super().__init__(name="NetBox", get_source_data=_get_reader(input_ref), *args, **kwargs)
    self.job = job
    self.sync = sync

    self.options = options
    self.diff_summary: DiffSummary = {}

    for name in GENERATOR_SETUP_MODULES:
        setup = __import__(name, fromlist=["setup"]).setup
        setup(self)
import_to_nautobot()

Import a NetBox export file into Nautobot.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
def import_to_nautobot(self) -> None:
    """Import a NetBox export file into Nautobot."""
    commited = False
    try:
        self._atomic_import()
        commited = True
    except _DryRunException:
        logger.warning("Dry-run mode, no data has been imported.")
    except _ValidationIssuesDetected:
        logger.warning("Data validation issues detected, no data has been imported.")

    if commited and self.options.update_paths:
        logger.info("Updating paths ...")
        call_command("trace_paths", no_input=True)
        logger.info(" ... Updating paths completed.")

    if self.options.summary:
        print_summary(self, self.diff_summary, self.options.field_mapping)
load()

Load data from NetBox.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
def load(self) -> None:
    """Load data from NetBox."""
    self.import_data()
    if self.options.fix_powerfeed_locations:
        fix_power_feed_locations(self)
    self.post_import()

NetBoxImporterOptions

Bases: NamedTuple

NetBox importer options.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
class NetBoxImporterOptions(NamedTuple):
    """NetBox importer options."""

    dry_run: bool = True
    bypass_data_validation: bool = False
    summary: bool = False
    field_mapping: bool = False
    update_paths: bool = False
    fix_powerfeed_locations: bool = False
    sitegroup_parent_always_region: bool = False

nautobot

Nautobot Adapter for NetBox Importer.

NautobotAdapter

Bases: NautobotAdapter

DiffSync adapter for Nautobot.

Source code in nautobot_netbox_importer/diffsync/adapters/nautobot.py
class NautobotAdapter(_NautobotAdapter):
    """DiffSync adapter for Nautobot."""

    def __init__(self, *args, job=None, sync=None, **kwargs):
        """Initialize Nautobot.

        Args:
            job (object, optional): Nautobot job. Defaults to None.
            sync (object, optional): Nautobot DiffSync. Defaults to None.
        """
        super().__init__(*args, **kwargs)
        self.job = job
        self.sync = sync
__init__(*args, job=None, sync=None, **kwargs)

Initialize Nautobot.

Parameters:

Name Type Description Default
job object

Nautobot job. Defaults to None.

None
sync object

Nautobot DiffSync. Defaults to None.

None
Source code in nautobot_netbox_importer/diffsync/adapters/nautobot.py
def __init__(self, *args, job=None, sync=None, **kwargs):
    """Initialize Nautobot.

    Args:
        job (object, optional): Nautobot job. Defaults to None.
        sync (object, optional): Nautobot DiffSync. Defaults to None.
    """
    super().__init__(*args, **kwargs)
    self.job = job
    self.sync = sync

netbox

NetBox to Nautobot Source Importer Definitions.

NetBoxAdapter

Bases: SourceAdapter

NetBox Source Adapter.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
class NetBoxAdapter(SourceAdapter):
    """NetBox Source Adapter."""

    # pylint: disable=keyword-arg-before-vararg
    def __init__(self, input_ref: _FileRef, options: NetBoxImporterOptions, job=None, sync=None, *args, **kwargs):
        """Initialize NetBox Source Adapter."""
        super().__init__(name="NetBox", get_source_data=_get_reader(input_ref), *args, **kwargs)
        self.job = job
        self.sync = sync

        self.options = options
        self.diff_summary: DiffSummary = {}

        for name in GENERATOR_SETUP_MODULES:
            setup = __import__(name, fromlist=["setup"]).setup
            setup(self)

    def load(self) -> None:
        """Load data from NetBox."""
        self.import_data()
        if self.options.fix_powerfeed_locations:
            fix_power_feed_locations(self)
        self.post_import()

    def import_to_nautobot(self) -> None:
        """Import a NetBox export file into Nautobot."""
        commited = False
        try:
            self._atomic_import()
            commited = True
        except _DryRunException:
            logger.warning("Dry-run mode, no data has been imported.")
        except _ValidationIssuesDetected:
            logger.warning("Data validation issues detected, no data has been imported.")

        if commited and self.options.update_paths:
            logger.info("Updating paths ...")
            call_command("trace_paths", no_input=True)
            logger.info(" ... Updating paths completed.")

        if self.options.summary:
            print_summary(self, self.diff_summary, self.options.field_mapping)

    @atomic
    def _atomic_import(self) -> None:
        self.load()

        diff = self.nautobot.sync_from(self)
        self.diff_summary = diff.summary()

        if self.nautobot.validation_issues and not self.options.bypass_data_validation:
            raise _ValidationIssuesDetected("Data validation issues detected, aborting the transaction.")

        if self.options.dry_run:
            raise _DryRunException("Aborting the transaction due to the dry-run mode.")
__init__(input_ref, options, job=None, sync=None, *args, **kwargs)

Initialize NetBox Source Adapter.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
def __init__(self, input_ref: _FileRef, options: NetBoxImporterOptions, job=None, sync=None, *args, **kwargs):
    """Initialize NetBox Source Adapter."""
    super().__init__(name="NetBox", get_source_data=_get_reader(input_ref), *args, **kwargs)
    self.job = job
    self.sync = sync

    self.options = options
    self.diff_summary: DiffSummary = {}

    for name in GENERATOR_SETUP_MODULES:
        setup = __import__(name, fromlist=["setup"]).setup
        setup(self)
import_to_nautobot()

Import a NetBox export file into Nautobot.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
def import_to_nautobot(self) -> None:
    """Import a NetBox export file into Nautobot."""
    commited = False
    try:
        self._atomic_import()
        commited = True
    except _DryRunException:
        logger.warning("Dry-run mode, no data has been imported.")
    except _ValidationIssuesDetected:
        logger.warning("Data validation issues detected, no data has been imported.")

    if commited and self.options.update_paths:
        logger.info("Updating paths ...")
        call_command("trace_paths", no_input=True)
        logger.info(" ... Updating paths completed.")

    if self.options.summary:
        print_summary(self, self.diff_summary, self.options.field_mapping)
load()

Load data from NetBox.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
def load(self) -> None:
    """Load data from NetBox."""
    self.import_data()
    if self.options.fix_powerfeed_locations:
        fix_power_feed_locations(self)
    self.post_import()
NetBoxImporterOptions

Bases: NamedTuple

NetBox importer options.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py
class NetBoxImporterOptions(NamedTuple):
    """NetBox importer options."""

    dry_run: bool = True
    bypass_data_validation: bool = False
    summary: bool = False
    field_mapping: bool = False
    update_paths: bool = False
    fix_powerfeed_locations: bool = False
    sitegroup_parent_always_region: bool = False

models

base

NetBox to Nautobot Base Models Mapping.

setup(adapter)

Map NetBox base models to Nautobot.

Source code in nautobot_netbox_importer/diffsync/models/base.py
def setup(adapter: SourceAdapter) -> None:
    """Map NetBox base models to Nautobot."""
    adapter.disable_model("sessions.session", "Nautobot has own sessions, sessions should never cross apps.")
    adapter.disable_model("admin.logentry", "Not directly used in Nautobot.")
    adapter.disable_model("users.userconfig", "May not have a 1 to 1 translation to Nautobot.")
    adapter.disable_model("auth.permission", "Handled via a Nautobot model and may not be a 1 to 1.")

    _setup_content_types(adapter)
    _setup_object_change(adapter)

    adapter.configure_model(
        "extras.status",
        identifiers=["name"],
        default_reference={
            "name": "Unknown",
            "content_types": [],
        },
    )
    adapter.configure_model("extras.role")
    adapter.configure_model(
        "extras.customfield",
        fields={
            "name": "key",
            "choices": _define_choices,
        },
    )
    adapter.configure_model(
        "extras.customfieldchoice",
        fields={
            "custom_field": "custom_field",
            "value": "value",
        },
    )
    adapter.configure_model(
        "extras.taggeditem",
        fields={
            "object_id": _define_tagged_object,
        },
    )
    adapter.configure_model(
        "auth.user",
        nautobot_content_type="users.user",
        identifiers=["username"],
        fields={
            "last_login": fields.disable("Should not be attempted to migrated"),
            "password": fields.disable("Should not be attempted to migrated"),
            "user_permissions": fields.disable("Permissions import are not implemented yet"),
        },
    )
    adapter.configure_model(
        "auth.group",
        identifiers=["name"],
        fields={
            "permissions": fields.disable("Permissions import are not implemented yet"),
        },
    )
    adapter.configure_model(
        "tenancy.tenant",
        fields={
            "group": "tenant_group",
        },
    )

circuits

NetBox to Nautobot Circuits Models Mapping.

setup(adapter)

Map NetBox circuits models to Nautobot.

Source code in nautobot_netbox_importer/diffsync/models/circuits.py
def setup(adapter: SourceAdapter) -> None:
    """Map NetBox circuits models to Nautobot."""
    adapter.configure_model(
        "circuits.circuit",
        fields={
            "type": "circuit_type",
            "termination_a": "circuit_termination_a",
            "termination_z": "circuit_termination_z",
        },
    )
    adapter.configure_model(
        "circuits.circuittermination",
        fields={
            "location": define_location,
        },
    )

dcim

NetBox to Nautobot DCIM Models Mapping.

fix_power_feed_locations(adapter)

Fix panel location to match rack location based on powerfeed.

Source code in nautobot_netbox_importer/diffsync/models/dcim.py
def fix_power_feed_locations(adapter: SourceAdapter) -> None:
    """Fix panel location to match rack location based on powerfeed."""
    region_wrapper = adapter.wrappers["dcim.region"]
    site_wrapper = adapter.wrappers["dcim.site"]
    location_wrapper = adapter.wrappers["dcim.location"]
    rack_wrapper = adapter.wrappers["dcim.rack"]
    panel_wrapper = adapter.wrappers["dcim.powerpanel"]

    diffsync_class = adapter.wrappers["dcim.powerfeed"].nautobot.diffsync_class

    for item in adapter.get_all(diffsync_class):
        rack_id = getattr(item, "rack_id", None)
        panel_id = getattr(item, "power_panel_id", None)
        if not (rack_id and panel_id):
            continue

        rack = rack_wrapper.get_or_create(rack_id)
        panel = panel_wrapper.get_or_create(panel_id)

        rack_location_uid = getattr(rack, "location_id", None)
        panel_location_uid = getattr(panel, "location_id", None)
        if rack_location_uid == panel_location_uid:
            continue

        if rack_location_uid:
            location_uid = rack_location_uid
            target = panel
            target_wrapper = panel_wrapper
        else:
            location_uid = panel_location_uid
            target = rack
            target_wrapper = rack_wrapper

        if not isinstance(location_uid, UUID):
            raise TypeError(f"Location UID must be UUID, got {type(location_uid)}")

        target.location_id = location_uid
        adapter.update(target)

        # Need to update references, to properly update `content_types` fields
        # References can be counted and removed, if needed
        if location_uid in region_wrapper.references:
            region_wrapper.add_reference(location_uid, target_wrapper)
        elif location_uid in site_wrapper.references:
            site_wrapper.add_reference(location_uid, target_wrapper)
        elif location_uid in location_wrapper.references:
            location_wrapper.add_reference(location_uid, target_wrapper)
        else:
            raise ValueError(f"Unknown location type {location_uid}")
setup(adapter)

Map NetBox DCIM models to Nautobot.

Source code in nautobot_netbox_importer/diffsync/models/dcim.py
def setup(adapter: SourceAdapter) -> None:
    """Map NetBox DCIM models to Nautobot."""
    adapter.disable_model("dcim.cablepath", "Recreated in Nautobot on signal when circuit termination is created")
    adapter.configure_model(
        "dcim.rackreservation",
        fields={
            "units": _define_units,
        },
    )
    adapter.configure_model(
        "dcim.rack",
        fields={
            "location": define_location,
            "role": fields.role(adapter, "dcim.rackrole"),
        },
    )
    adapter.configure_model(
        "dcim.cable",
        fields={
            "termination_a": "termination_a",
            "termination_b": "termination_b",
        },
    )
    adapter.configure_model(
        "dcim.cabletermination",
        extend_content_type="dcim.cable",
        pre_import=_pre_import_cable_termination,
        fields={
            "termination_a": "termination_a",
            "termination_b": "termination_b",
        },
    )
    adapter.configure_model(
        "dcim.interface",
        fields={
            "status": "status",
            "parent": "parent_interface",
        },
    )
    manufacturer = adapter.configure_model(
        "dcim.manufacturer",
        default_reference={
            "id": "Unknown",
            "name": "Unknown",
        },
    )
    adapter.configure_model(
        "dcim.devicetype",
        fields={
            "front_image": fields.disable("Import does not contain images"),
            "rear_image": fields.disable("Import does not contain images"),
            "color": "color",
        },
        default_reference={
            "id": "Unknown",
            "manufacturer": manufacturer.get_default_reference_uid(),
            "model": "Unknown",
        },
    )
    adapter.configure_model(
        "dcim.devicerole",
        nautobot_content_type="extras.role",
    )
    adapter.configure_model(
        "dcim.device",
        fields={
            "location": define_location,
            "device_role": fields.role(adapter, "dcim.devicerole"),
            "role": fields.role(adapter, "dcim.devicerole"),
        },
    )
    adapter.configure_model(
        "dcim.powerpanel",
        fields={
            "location": define_location,
        },
    )
    adapter.configure_model(
        "dcim.frontporttemplate",
        fields={
            "rear_port": "rear_port_template",
        },
    )
    adapter.configure_model(
        "dcim.poweroutlettemplate",
        fields={
            "power_port": "power_port_template",
        },
    )

ipam

NetBox to Nautobot IPAM Models Mapping.

setup(adapter)

Map NetBox IPAM models to Nautobot.

Source code in nautobot_netbox_importer/diffsync/models/ipam.py
def setup(adapter: SourceAdapter) -> None:
    """Map NetBox IPAM models to Nautobot."""
    ipaddress = adapter.configure_model(
        "ipam.ipaddress",
        fields={
            "role": fields.role(adapter, "ipam.role"),
        },
    )
    ipaddress.nautobot.set_instance_defaults(namespace=get_default_namespace())
    adapter.configure_model(
        "ipam.prefix",
        fields={
            "location": define_location,
            "role": fields.role(adapter, "ipam.role"),
        },
    )
    adapter.configure_model(
        "ipam.aggregate",
        "ipam.prefix",
        fields={
            "status": "status",
        },
    )
    adapter.configure_model(
        "ipam.vlan",
        fields={
            "group": "vlan_group",
            "location": define_location,
            "role": fields.role(adapter, "ipam.role"),
        },
    )

locations

NetBox Specific Locations handling.

define_location(field)

Define location field for NetBox importer.

Source code in nautobot_netbox_importer/diffsync/models/locations.py
def define_location(field: SourceField) -> None:
    """Define location field for NetBox importer."""
    field.set_nautobot_field(field.name)
    field.handle_sibling("site")
    field.handle_sibling("region")

    wrapper = field.wrapper

    location_wrapper = wrapper.adapter.wrappers["dcim.location"]
    site_wrapper = wrapper.adapter.wrappers["dcim.site"]
    region_wrapper = wrapper.adapter.wrappers["dcim.region"]

    def importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        location = source.get(field.name, None)
        site = source.get("site", None)
        region = source.get("region", None)

        # Location is the most specific, then site, the last is region
        if location:
            result = location_wrapper.get_pk_from_uid(location)
            location_wrapper.add_reference(result, wrapper)
        elif site:
            result = site_wrapper.get_pk_from_uid(site)
            site_wrapper.add_reference(result, wrapper)
        elif region:
            result = region_wrapper.get_pk_from_uid(region)
            region_wrapper.add_reference(result, wrapper)
        else:
            return

        setattr(target, field.nautobot.name, result)

    field.set_importer(importer)
setup(adapter)

Setup locations for NetBox importer.

Source code in nautobot_netbox_importer/diffsync/models/locations.py
def setup(adapter: SourceAdapter) -> None:
    """Setup locations for NetBox importer."""
    options = getattr(adapter, "options", {})
    sitegroup_parent_always_region = getattr(options, "sitegroup_parent_always_region", False)

    location_type_wrapper = adapter.configure_model("dcim.locationtype")

    region_type_uid = location_type_wrapper.cache_record(
        {
            "id": "Region",
            "name": "Region",
            "nestable": True,
            "content_types": None,
        }
    )
    site_type_uid = location_type_wrapper.cache_record(
        {
            "id": "Site",
            "name": "Site",
            "nestable": False,
            "parent": region_type_uid,
        }
    )
    location_type_uid = location_type_wrapper.cache_record(
        {
            "id": "Location",
            "name": "Location",
            "nestable": True,
            "parent": site_type_uid,
        }
    )

    adapter.configure_model(
        "dcim.sitegroup",
        nautobot_content_type="dcim.locationtype",
        fields={
            "parent": fields.constant(region_type_uid) if sitegroup_parent_always_region else "parent",
            "nestable": fields.constant(True),
        },
    )

    adapter.configure_model(
        "dcim.region",
        nautobot_content_type="dcim.location",
        fields={
            "status": "status",
            "location_type": fields.constant(region_type_uid),
        },
    )

    adapter.configure_model(
        "dcim.site",
        nautobot_content_type="dcim.location",
        fields={
            "region": _define_site_region,
            "group": _define_site_group,
        },
    )

    adapter.configure_model(
        "dcim.location",
        fields={
            "status": "status",
            "location_type": fields.constant(location_type_uid),
            "parent": _define_location_parent,
        },
    )

    for name in ["region", "site", "location"]:
        wrapper = adapter.wrappers[f"dcim.{name}"]
        location_type_wrapper.add_reference(location_type_wrapper.get_pk_from_uid(name.capitalize()), wrapper)
        wrapper.set_references_forwarding("dcim.locationtype", "location_type_id")

virtualization

NetBox to Nautobot Virtualization Models Mapping.

setup(adapter)

Map NetBox virtualization models to Nautobot.

Source code in nautobot_netbox_importer/diffsync/models/virtualization.py
def setup(adapter: SourceAdapter) -> None:
    """Map NetBox virtualization models to Nautobot."""
    adapter.configure_model(
        "virtualization.cluster",
        fields={
            "type": "cluster_type",
            "group": "cluster_group",
            "location": define_location,
        },
    )
    adapter.configure_model(
        "virtualization.virtualmachine",
        fields={
            "role": fields.role(adapter, "dcim.devicerole"),
        },
    )
    adapter.configure_model(
        "virtualization.vminterface",
        fields={
            "status": "status",
            "parent": "parent_interface",
        },
    )