Skip to content

Nautobot NetBox Importer Package

nautobot_netbox_importer

App declaration for nautobot_netbox_importer.

NautobotNetboxImporterConfig

Bases: NautobotAppConfig

App configuration for the nautobot_netbox_importer app.

Source code in nautobot_netbox_importer/__init__.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class NautobotNetboxImporterConfig(NautobotAppConfig):
    """App configuration for the nautobot_netbox_importer app."""

    name = "nautobot_netbox_importer"
    verbose_name = "Nautobot NetBox Importer"
    version = __version__
    author = "Network to Code, LLC"
    description = "Data importer from NetBox 3.x to Nautobot 2.x."
    base_url = "netbox-importer"
    required_settings = []
    min_version = "2.0.6"
    max_version = "2.9999"
    default_settings = {}
    caching_config = {}

base

Base types for the Nautobot Importer.

register_generator_setup(module)

Register adapter setup function.

This function must be called before the adapter is used and containing module can't import anything from Nautobot.

Source code in nautobot_netbox_importer/base.py 
26
27
28
29
30
31
32
def register_generator_setup(module: str) -> None:
    """Register adapter setup function.

    This function must be called before the adapter is used and containing module can't import anything from Nautobot.
    """
    if module not in GENERATOR_SETUP_MODULES:
        GENERATOR_SETUP_MODULES.append(module)

command_utils

Utility functions and classes for nautobot_netbox_importer.

LogRenderer

Class for rendering structured logs to the console in a human-readable format.

Example

19:48:19 Apparent duplicate object encountered? duplicate_id: {'group': None, 'name': 'CR02.CHI_ORDMGMT', 'site': {'name': 'CHI01'}, 'vid': 1000} model: vlan pk_1: 3baf142d-dd90-4379-a048-3bbbcc9c799c pk_2: cba19791-4d59-4ddd-a5c9-d969ec3ed2ba

Source code in nautobot_netbox_importer/command_utils.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class LogRenderer:  # pylint: disable=too-few-public-methods
    """Class for rendering structured logs to the console in a human-readable format.

    Example:
        19:48:19 Apparent duplicate object encountered?
          duplicate_id:
            {'group': None,
            'name': 'CR02.CHI_ORDMGMT',
            'site': {'name': 'CHI01'},
            'vid': 1000}
          model: vlan
          pk_1: 3baf142d-dd90-4379-a048-3bbbcc9c799c
          pk_2: cba19791-4d59-4ddd-a5c9-d969ec3ed2ba
    """

    def __call__(
        self,
        logger: structlog.types.WrappedLogger,
        name: str,
        event_dict: structlog.types.EventDict,
    ) -> str:
        """Render the given event_dict to a string."""
        sio = StringIO()

        timestamp = event_dict.pop("timestamp", None)
        if timestamp is not None:
            sio.write(f"{colorama.Style.DIM}{timestamp}{colorama.Style.RESET_ALL} ")

        level = event_dict.pop("level", None)
        if level is not None:
            if level in ("warning", "error", "critical"):
                sio.write(f"{colorama.Fore.RED}{level:<9}{colorama.Style.RESET_ALL}")
            else:
                sio.write(f"{level:<9}")

        event = event_dict.pop("event", None)
        sio.write(f"{colorama.Style.BRIGHT}{event}{colorama.Style.RESET_ALL}")

        for key, value in event_dict.items():
            if isinstance(value, dict):
                # We could use json.dumps() here instead of pprint.pformat,
                # but I find pprint to be a bit more compact while still readable.
                rendered_dict = pprint.pformat(value)
                if len(rendered_dict.splitlines()) > 50:
                    rendered_dict = "\n".join(rendered_dict.splitlines()[:50]) + "\n...}"
                value = "\n" + textwrap.indent(rendered_dict, "    ")
            sio.write(
                f"\n  {colorama.Fore.CYAN}{key}{colorama.Style.RESET_ALL}: "
                f"{colorama.Fore.MAGENTA}{value}{colorama.Style.RESET_ALL}"
            )

        return sio.getvalue()
__call__(logger, name, event_dict)

Render the given event_dict to a string.

Source code in nautobot_netbox_importer/command_utils.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __call__(
    self,
    logger: structlog.types.WrappedLogger,
    name: str,
    event_dict: structlog.types.EventDict,
) -> str:
    """Render the given event_dict to a string."""
    sio = StringIO()

    timestamp = event_dict.pop("timestamp", None)
    if timestamp is not None:
        sio.write(f"{colorama.Style.DIM}{timestamp}{colorama.Style.RESET_ALL} ")

    level = event_dict.pop("level", None)
    if level is not None:
        if level in ("warning", "error", "critical"):
            sio.write(f"{colorama.Fore.RED}{level:<9}{colorama.Style.RESET_ALL}")
        else:
            sio.write(f"{level:<9}")

    event = event_dict.pop("event", None)
    sio.write(f"{colorama.Style.BRIGHT}{event}{colorama.Style.RESET_ALL}")

    for key, value in event_dict.items():
        if isinstance(value, dict):
            # We could use json.dumps() here instead of pprint.pformat,
            # but I find pprint to be a bit more compact while still readable.
            rendered_dict = pprint.pformat(value)
            if len(rendered_dict.splitlines()) > 50:
                rendered_dict = "\n".join(rendered_dict.splitlines()[:50]) + "\n...}"
            value = "\n" + textwrap.indent(rendered_dict, "    ")
        sio.write(
            f"\n  {colorama.Fore.CYAN}{key}{colorama.Style.RESET_ALL}: "
            f"{colorama.Fore.MAGENTA}{value}{colorama.Style.RESET_ALL}"
        )

    return sio.getvalue()

enable_logging(verbosity=0, color=None)

Set up structlog (as used by DiffSync) to log messages for this command.

Source code in nautobot_netbox_importer/command_utils.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def enable_logging(verbosity=0, color=None):
    """Set up structlog (as used by DiffSync) to log messages for this command."""
    if color is None:
        # Let colorama decide whether or not to strip out color codes
        colorama.init()
    else:
        # Force colors or non-colors, as specified
        colorama.init(strip=not color)

    structlog.configure(
        processors=[
            structlog.stdlib.add_log_level,
            structlog.processors.TimeStamper(fmt="%H:%M:%S"),
            LogRenderer(),
        ],
        context_class=dict,
        # Logging levels aren't very granular, so we adjust the log level based on *half* the verbosity level:
        # Verbosity     Logging level
        # 0             30 (WARNING)
        # 1-2           20 (INFO)
        # 3+            10 (DEBUG)
        wrapper_class=structlog.make_filtering_bound_logger(10 * (3 - ((verbosity + 1) // 2))),
        cache_logger_on_first_use=True,
    )

initialize_logger(options)

Initialize logger instance.

Source code in nautobot_netbox_importer/command_utils.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def initialize_logger(options):
    """Initialize logger instance."""
    # Default of None means to use colorama's autodetection to determine whether or not to use color
    color = None
    if options.get("force_color"):
        color = True
    if options.get("no_color"):
        color = False

    enable_logging(verbosity=options["verbosity"], color=color)
    return structlog.get_logger(), color

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 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
class NautobotAdapter(_NautobotAdapter):
    """DiffSync adapter for Nautobot."""

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

        Args:
            *args (tuple): Arguments to be passed to the parent class.
            job (object, optional): Nautobot job. Defaults to None.
            sync (object, optional): Nautobot DiffSync. Defaults to None.
            **kwargs (dict): Additional arguments to be passed to the parent class.
        """
        super().__init__(*args, **kwargs)
        self.job = job
        self.sync = sync
__init__(*args, job=None, sync=None, **kwargs)

Initialize Nautobot.

Parameters:

Name Type Description Default
*args tuple

Arguments to be passed to the parent class.

 ()
job object

Nautobot job. Defaults to None.

 None
sync object

Nautobot DiffSync. Defaults to None.

 None
**kwargs dict

Additional arguments to be passed to the parent class.

 {}
Source code in nautobot_netbox_importer/diffsync/adapters/nautobot.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
def __init__(self, *args, job=None, sync=None, **kwargs):
    """Initialize Nautobot.

    Args:
        *args (tuple): Arguments to be passed to the parent class.
        job (object, optional): Nautobot job. Defaults to None.
        sync (object, optional): Nautobot DiffSync. Defaults to None.
        **kwargs (dict): Additional arguments to be passed to the parent class.
    """
    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 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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

        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)
        if self.options.unrack_zero_uheight_devices:
            unrack_zero_uheight_devices(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 _ImporterIssuesDetected:
            logger.warning("Importer 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.print_summary:
            self.summary.print()

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

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

        if self.options.save_json_summary_path:
            self.summary.dump(self.options.save_json_summary_path, output_format="json")
        if self.options.save_text_summary_path:
            self.summary.dump(self.options.save_text_summary_path, output_format="text")

        has_issues = any(True for item in self.summary.nautobot if item.issues)
        if has_issues and not self.options.bypass_data_validation:
            raise _ImporterIssuesDetected("Importer 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 
72
73
74
75
76
77
78
79
80
81
82
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

    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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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 _ImporterIssuesDetected:
        logger.warning("Importer 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.print_summary:
        self.summary.print()
load()

Load data from NetBox.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py 
84
85
86
87
88
89
90
91
def load(self) -> None:
    """Load data from NetBox."""
    self.import_data()
    if self.options.fix_powerfeed_locations:
        fix_power_feed_locations(self)
    if self.options.unrack_zero_uheight_devices:
        unrack_zero_uheight_devices(self)
    self.post_import()
NetBoxImporterOptions

Bases: NamedTuple

NetBox importer options.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py 
51
52
53
54
55
56
57
58
59
60
61
62
class NetBoxImporterOptions(NamedTuple):
    """NetBox importer options."""

    dry_run: bool = True
    bypass_data_validation: bool = False
    print_summary: bool = False
    update_paths: bool = False
    fix_powerfeed_locations: bool = False
    sitegroup_parent_always_region: bool = False
    unrack_zero_uheight_devices: bool = True
    save_json_summary_path: str = ""
    save_text_summary_path: str = ""
nautobot

Nautobot Adapter for NetBox Importer.

NautobotAdapter

Bases: NautobotAdapter

DiffSync adapter for Nautobot.

Source code in nautobot_netbox_importer/diffsync/adapters/nautobot.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
class NautobotAdapter(_NautobotAdapter):
    """DiffSync adapter for Nautobot."""

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

        Args:
            *args (tuple): Arguments to be passed to the parent class.
            job (object, optional): Nautobot job. Defaults to None.
            sync (object, optional): Nautobot DiffSync. Defaults to None.
            **kwargs (dict): Additional arguments to be passed to the parent class.
        """
        super().__init__(*args, **kwargs)
        self.job = job
        self.sync = sync
__init__(*args, job=None, sync=None, **kwargs)

Initialize Nautobot.

Parameters:

Name Type Description Default
*args tuple

Arguments to be passed to the parent class.

 ()
job object

Nautobot job. Defaults to None.

 None
sync object

Nautobot DiffSync. Defaults to None.

 None
**kwargs dict

Additional arguments to be passed to the parent class.

 {}
Source code in nautobot_netbox_importer/diffsync/adapters/nautobot.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
def __init__(self, *args, job=None, sync=None, **kwargs):
    """Initialize Nautobot.

    Args:
        *args (tuple): Arguments to be passed to the parent class.
        job (object, optional): Nautobot job. Defaults to None.
        sync (object, optional): Nautobot DiffSync. Defaults to None.
        **kwargs (dict): Additional arguments to be passed to the parent class.
    """
    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 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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

        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)
        if self.options.unrack_zero_uheight_devices:
            unrack_zero_uheight_devices(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 _ImporterIssuesDetected:
            logger.warning("Importer 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.print_summary:
            self.summary.print()

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

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

        if self.options.save_json_summary_path:
            self.summary.dump(self.options.save_json_summary_path, output_format="json")
        if self.options.save_text_summary_path:
            self.summary.dump(self.options.save_text_summary_path, output_format="text")

        has_issues = any(True for item in self.summary.nautobot if item.issues)
        if has_issues and not self.options.bypass_data_validation:
            raise _ImporterIssuesDetected("Importer 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 
72
73
74
75
76
77
78
79
80
81
82
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

    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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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 _ImporterIssuesDetected:
        logger.warning("Importer 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.print_summary:
        self.summary.print()
load()

Load data from NetBox.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py 
84
85
86
87
88
89
90
91
def load(self) -> None:
    """Load data from NetBox."""
    self.import_data()
    if self.options.fix_powerfeed_locations:
        fix_power_feed_locations(self)
    if self.options.unrack_zero_uheight_devices:
        unrack_zero_uheight_devices(self)
    self.post_import()
NetBoxImporterOptions

Bases: NamedTuple

NetBox importer options.

Source code in nautobot_netbox_importer/diffsync/adapters/netbox.py 
51
52
53
54
55
56
57
58
59
60
61
62
class NetBoxImporterOptions(NamedTuple):
    """NetBox importer options."""

    dry_run: bool = True
    bypass_data_validation: bool = False
    print_summary: bool = False
    update_paths: bool = False
    fix_powerfeed_locations: bool = False
    sitegroup_parent_always_region: bool = False
    unrack_zero_uheight_devices: bool = True
    save_json_summary_path: str = ""
    save_text_summary_path: str = ""

models

NetBox Importer DiffSync 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 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
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)

    adapter.configure_model(
        "extras.Status",
        identifiers=["name"],
        default_reference={
            "name": "Unknown",
        },
    )
    adapter.configure_model("extras.role")
    adapter.configure_model(
        "extras.tag",
        fields={
            "object_types": "content_types",
        },
    )
    adapter.configure_model(
        "extras.TaggedItem",
        fields={
            "object_id": _define_tagged_object,
        },
    )
    adapter.configure_model(
        # pylint: disable=hard-coded-auth-user
        "auth.User",
        nautobot_content_type="users.User",
        identifiers=["username"],
        fields={
            "last_login": fields.disable("Should not be attempted to migrate"),
            "password": fields.disable("Should not be attempted to migrate"),
            "user_permissions": fields.disable("Permissions import is not implemented yet"),
        },
    )
    adapter.configure_model(
        "auth.Group",
        identifiers=["name"],
        fields={
            "permissions": fields.disable("Permissions import is not implemented yet"),
        },
    )
    adapter.configure_model(
        "tenancy.Tenant",
        fields={
            "group": "tenant_group",
        },
    )
    adapter.configure_model(
        "extras.JournalEntry",
        nautobot_content_type="extras.Note",
    )
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 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
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,
        },
    )
custom_fields

NetBox to Nautobot Custom Fields Models Mapping.

setup(adapter)

Map NetBox custom fields to Nautobot.

Source code in nautobot_netbox_importer/diffsync/models/custom_fields.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def setup(adapter: SourceAdapter) -> None:
    """Map NetBox custom fields to Nautobot."""
    choice_sets = {}

    def create_choice_set(source: RecordData, importer_pass: ImporterPass) -> PreImportResult:
        if importer_pass == ImporterPass.DEFINE_STRUCTURE:
            choice_sets[source.get("id")] = [
                *_convert_choices(source.get("base_choices")),
                *_convert_choices(source.get("extra_choices")),
            ]

        return PreImportResult.USE_RECORD

    def define_choice_set(field: SourceField) -> None:
        def choices_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            choice_set = source.get(field.name, None)
            if choice_set in EMPTY_VALUES:
                return

            choices = choice_sets.get(choice_set, None)
            if not choices:
                raise ValueError(f"Choice set {choice_set} not found")

            create_choices(choices, getattr(target, "id"))

        field.set_importer(choices_importer, nautobot_name=None)

    def define_choices(field: SourceField) -> None:
        def choices_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            choices = _convert_choices(source.get(field.name, None))
            if choices in EMPTY_VALUES:
                return

            if not isinstance(choices, list):
                raise ValueError(f"Choices must be a list of strings, got {type(choices)}")

            create_choices(choices, getattr(target, "id"))

        field.set_importer(choices_importer, nautobot_name=None)

    def create_choices(choices: list, custom_field_uid: Uid) -> None:
        for choice in choices:
            choices_wrapper.import_record(
                {
                    "id": choice,
                    "custom_field": custom_field_uid,
                    "value": choice,
                },
            )

    # Defined in NetBox but not in Nautobot
    adapter.configure_model(
        "extras.CustomFieldChoiceSet",
        pre_import=create_choice_set,
    )

    adapter.configure_model(
        "extras.CustomField",
        fields={
            "name": "key",
            "label": fields.default("Empty Label"),
            "type": fields.fallback(value="text"),
            # NetBox<3.6
            "choices": define_choices,
            # NetBox>=3.6
            "choice_set": define_choice_set,
        },
    )

    choices_wrapper = adapter.configure_model(
        "extras.CustomFieldChoice",
        fields={
            "custom_field": "custom_field",
            "value": "value",
        },
    )
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 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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:
            target_wrapper.add_reference(region_wrapper, location_uid)
        elif location_uid in site_wrapper.references:
            target_wrapper.add_reference(site_wrapper, location_uid)
        elif location_uid in location_wrapper.references:
            target_wrapper.add_reference(location_wrapper, location_uid)
        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 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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={
            # Set the definition of the `units` field
            "units": _define_units,
        },
    )
    adapter.configure_model(
        "dcim.rack",
        fields={
            "location": define_location,
            "role": fields.role(adapter, "dcim.rackrole"),
        },
    )
    adapter.configure_model("dcim.cable")
    adapter.configure_model(
        "dcim.cabletermination",
        extend_content_type="dcim.cable",
        pre_import=_pre_import_cable_termination,
    )
    adapter.configure_model(
        "dcim.interface",
        fields={
            "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"),
        },
        default_reference={
            "id": "Unknown",
            "manufacturer": manufacturer.get_default_reference_uid(),
            "model": "Unknown",
        },
    )
    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",
        },
    )
unrack_zero_uheight_devices(adapter)

Unrack devices with 0U height.

Source code in nautobot_netbox_importer/diffsync/models/dcim.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def unrack_zero_uheight_devices(adapter: SourceAdapter) -> None:
    """Unrack devices with 0U height."""
    device_wrapper = adapter.wrappers["dcim.device"]
    device_type_wrapper = adapter.wrappers["dcim.devicetype"]

    # Find all device types with 0U height
    device_type_ids = set(
        getattr(item, "id")
        for item in adapter.get_all(device_type_wrapper.nautobot.diffsync_class)
        if getattr(item, "u_height", 0) == 0
    )

    if not device_type_ids:
        return

    # Update all devices with matching device type, clean `position` field.
    position = device_wrapper.fields["position"]
    for item in adapter.get_all(device_wrapper.nautobot.diffsync_class):
        if getattr(item, "position", None) and getattr(item, "device_type_id") in device_type_ids:
            position.set_nautobot_value(item, None)
            adapter.update(item)
            position.add_issue("Unracked", "Device unracked due to 0U height", item)
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
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",
        nautobot_content_type="ipam.prefix",
    )
    adapter.configure_model(
        "ipam.vlan",
        fields={
            "group": "vlan_group",
            "location": define_location,
            "role": fields.role(adapter, "ipam.role"),
        },
    )
    adapter.configure_model(
        "ipam.FHRPGroup",
        nautobot_content_type="dcim.InterfaceRedundancyGroup",
        fields={
            "protocol": fields.fallback(callback=_fhrp_protocol_fallback),
        },
    )
locations

NetBox Specific Locations handling.

define_location(field)

Define location field for NetBox importer.

Source code in nautobot_netbox_importer/diffsync/models/locations.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def define_location(field: SourceField) -> None:
    """Define location field for NetBox importer."""
    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 location_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)
            wrapper.add_reference(location_wrapper, result)
        elif site:
            result = site_wrapper.get_pk_from_uid(site)
            wrapper.add_reference(site_wrapper, result)
        elif region:
            result = region_wrapper.get_pk_from_uid(region)
            wrapper.add_reference(region_wrapper, result)
        else:
            return

        field.set_nautobot_value(target, result)

    field.set_importer(location_importer)
    field.handle_sibling("site", field.nautobot.name)
    field.handle_sibling("region", field.nautobot.name)
setup(adapter)

Setup locations for NetBox importer.

Source code in nautobot_netbox_importer/diffsync/models/locations.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def setup(adapter: SourceAdapter) -> None:
    """Setup locations for NetBox importer."""

    def forward_references(wrapper: SourceModelWrapper, references: SourceReferences) -> None:
        """Forward references to Location, Site and Region instance to their LocationType to fill `content_types`."""
        for uid, wrappers in references.items():
            instance = wrapper.get_or_create(uid, fail_missing=True)
            location_type_uid = getattr(instance, "location_type_id")
            for item in wrappers:
                item.add_reference(location_type_wrapper, location_type_uid)

    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,
        }
    )
    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",
        forward_references=forward_references,
        fields={
            "location_type": fields.constant(region_type_uid),
        },
    )

    adapter.configure_model(
        "dcim.Site",
        nautobot_content_type="dcim.Location",
        forward_references=forward_references,
        fields={
            "region": fields.relation("dcim.Region", "parent"),
            "group": _define_site_group,
        },
    )

    adapter.configure_model(
        "dcim.Location",
        forward_references=forward_references,
        fields={
            "location_type": fields.constant(location_type_uid),
            "parent": _define_location_parent,
        },
    )
object_change

NetBox to Nautobot Object Change Model Mapping.

setup(adapter)

Map NetBox object change to Nautobot.

Source code in nautobot_netbox_importer/diffsync/models/object_change.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def setup(adapter: SourceAdapter) -> None:
    """Map NetBox object change to Nautobot."""

    def skip_disabled_object_types(source: RecordData, importer_pass: ImporterPass) -> PreImportResult:
        """Disabled object types are not in Nautobot and should be skipped."""
        if importer_pass != ImporterPass.IMPORT_DATA:
            return PreImportResult.USE_RECORD
        object_type = source.get("changed_object_type", None)
        wrapper = adapter.get_or_create_wrapper(object_type)
        return PreImportResult.SKIP_RECORD if wrapper.disable_reason else PreImportResult.USE_RECORD

    adapter.configure_model(
        "extras.ObjectChange",
        pre_import=skip_disabled_object_types,
        disable_related_reference=True,
        fields={
            "postchange_data": "object_data",
            # TBD: This should be defined on Nautobot side
            "time": fields.force(),
        },
    )
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 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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={
            "parent": "parent_interface",
        },
    )

generator

Generic Nautobot Import Library using DiffSync.

DiffSyncBaseModel

Bases: DiffSyncModel

Base class for all DiffSync models.

Source code in nautobot_netbox_importer/generator/nautobot.py 
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
class DiffSyncBaseModel(DiffSyncModel):
    """Base class for all DiffSync models."""

    _wrapper: NautobotModelWrapper

    @classmethod
    def create(cls, diffsync: DiffSync, ids: dict, attrs: dict) -> Optional["DiffSyncBaseModel"]:
        """Create this model instance, both in Nautobot and in DiffSync."""
        instance = cls._wrapper.model(**cls._wrapper.constructor_kwargs, **ids)

        cls._wrapper.save_nautobot_instance(instance, attrs)

        return super().create(diffsync, ids, attrs)

    def update(self, attrs: dict) -> Optional["DiffSyncBaseModel"]:
        """Update this model instance, both in Nautobot and in DiffSync."""
        uid = getattr(self, self._wrapper.pk_field.name, None)
        if not uid:
            raise NotImplementedError("Cannot update model without pk")

        model = self._wrapper.model
        filter_kwargs = {self._wrapper.pk_field.name: uid}
        instance = model.objects.get(**filter_kwargs)
        self._wrapper.save_nautobot_instance(instance, attrs)

        return super().update(attrs)
create(diffsync, ids, attrs) classmethod

Create this model instance, both in Nautobot and in DiffSync.

Source code in nautobot_netbox_importer/generator/nautobot.py 
573
574
575
576
577
578
579
580
@classmethod
def create(cls, diffsync: DiffSync, ids: dict, attrs: dict) -> Optional["DiffSyncBaseModel"]:
    """Create this model instance, both in Nautobot and in DiffSync."""
    instance = cls._wrapper.model(**cls._wrapper.constructor_kwargs, **ids)

    cls._wrapper.save_nautobot_instance(instance, attrs)

    return super().create(diffsync, ids, attrs)
update(attrs)

Update this model instance, both in Nautobot and in DiffSync.

Source code in nautobot_netbox_importer/generator/nautobot.py 
582
583
584
585
586
587
588
589
590
591
592
593
def update(self, attrs: dict) -> Optional["DiffSyncBaseModel"]:
    """Update this model instance, both in Nautobot and in DiffSync."""
    uid = getattr(self, self._wrapper.pk_field.name, None)
    if not uid:
        raise NotImplementedError("Cannot update model without pk")

    model = self._wrapper.model
    filter_kwargs = {self._wrapper.pk_field.name: uid}
    instance = model.objects.get(**filter_kwargs)
    self._wrapper.save_nautobot_instance(instance, attrs)

    return super().update(attrs)

ImporterPass

Bases: Enum

Importer Pass.

Source code in nautobot_netbox_importer/generator/source.py 
89
90
91
92
93
class ImporterPass(Enum):
    """Importer Pass."""

    DEFINE_STRUCTURE = 1
    IMPORT_DATA = 2

InvalidChoiceValueIssue

Bases: SourceFieldImporterIssue

Raised when an invalid choice value is encountered.

Source code in nautobot_netbox_importer/generator/source.py 
71
72
73
74
75
76
77
78
79
class InvalidChoiceValueIssue(SourceFieldImporterIssue):
    """Raised when an invalid choice value is encountered."""

    def __init__(self, field: "SourceField", value: Any, replacement: Any = NOTHING):
        """Initialize the exception."""
        message = f"Invalid choice value: `{value}`"
        if replacement is not NOTHING:
            message += f", replaced with `{replacement}`"
        super().__init__(message, field)
__init__(field, value, replacement=NOTHING)

Initialize the exception.

Source code in nautobot_netbox_importer/generator/source.py 
74
75
76
77
78
79
def __init__(self, field: "SourceField", value: Any, replacement: Any = NOTHING):
    """Initialize the exception."""
    message = f"Invalid choice value: `{value}`"
    if replacement is not NOTHING:
        message += f", replaced with `{replacement}`"
    super().__init__(message, field)

NautobotAdapter

Bases: BaseAdapter

Nautobot DiffSync Adapter.

Source code in nautobot_netbox_importer/generator/nautobot.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class NautobotAdapter(BaseAdapter):
    """Nautobot DiffSync Adapter."""

    def __init__(self, *args, **kwargs):
        """Initialize the adapter."""
        super().__init__("Nautobot", *args, **kwargs)
        self.wrappers: Dict[ContentTypeStr, NautobotModelWrapper] = {}

    def get_or_create_wrapper(self, content_type: ContentTypeStr) -> "NautobotModelWrapper":
        """Get or create a Nautobot model wrapper."""
        if content_type in self.wrappers:
            return self.wrappers[content_type]

        return NautobotModelWrapper(self, content_type)
__init__(*args, **kwargs)

Initialize the adapter.

Source code in nautobot_netbox_importer/generator/nautobot.py 
125
126
127
128
def __init__(self, *args, **kwargs):
    """Initialize the adapter."""
    super().__init__("Nautobot", *args, **kwargs)
    self.wrappers: Dict[ContentTypeStr, NautobotModelWrapper] = {}
get_or_create_wrapper(content_type)

Get or create a Nautobot model wrapper.

Source code in nautobot_netbox_importer/generator/nautobot.py 
130
131
132
133
134
135
def get_or_create_wrapper(self, content_type: ContentTypeStr) -> "NautobotModelWrapper":
    """Get or create a Nautobot model wrapper."""
    if content_type in self.wrappers:
        return self.wrappers[content_type]

    return NautobotModelWrapper(self, content_type)

PreImportResult

Bases: Enum

Pre Import Response.

Source code in nautobot_netbox_importer/generator/source.py 
 96
 97
 98
 99
100
class PreImportResult(Enum):
    """Pre Import Response."""

    SKIP_RECORD = False
    USE_RECORD = True

SourceAdapter

Bases: BaseAdapter

Source DiffSync Adapter.

Source code in nautobot_netbox_importer/generator/source.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
class SourceAdapter(BaseAdapter):
    """Source DiffSync Adapter."""

    def __init__(
        self,
        get_source_data: SourceDataGenerator,
        *args,
        nautobot: Optional[NautobotAdapter] = None,
        logger=None,
        **kwargs,
    ):
        """Initialize the SourceAdapter."""
        super().__init__(*args, **kwargs)

        self.get_source_data = get_source_data
        self.wrappers: OrderedDict[ContentTypeStr, SourceModelWrapper] = OrderedDict()
        self.nautobot = nautobot or NautobotAdapter()
        self.content_type_ids_mapping: Dict[int, SourceModelWrapper] = {}
        self.logger = logger or default_logger
        self.summary = ImportSummary()

        # From Nautobot to Source content type mapping
        # When multiple source content types are mapped to the single nautobot content type, mapping is set to `None`
        self._content_types_back_mapping: Dict[ContentTypeStr, Optional[ContentTypeStr]] = {}

    # pylint: disable=too-many-arguments,too-many-branches,too-many-locals
    def configure_model(
        self,
        content_type: ContentTypeStr,
        nautobot_content_type: ContentTypeStr = "",
        extend_content_type: ContentTypeStr = "",
        identifiers: Optional[Iterable[FieldName]] = None,
        fields: Optional[Mapping[FieldName, SourceFieldDefinition]] = None,
        default_reference: Optional[RecordData] = None,
        flags: Optional[DiffSyncModelFlags] = None,
        nautobot_flags: Optional[DiffSyncModelFlags] = None,
        pre_import: Optional[PreImport] = None,
        disable_related_reference: Optional[bool] = None,
        forward_references: Optional[ForwardReferences] = None,
    ) -> "SourceModelWrapper":
        """Create if not exist and configure a wrapper for a given source content type.

        Create Nautobot content type wrapper as well.
        """
        content_type = content_type.lower()
        nautobot_content_type = nautobot_content_type.lower()
        extend_content_type = extend_content_type.lower()

        if extend_content_type:
            if nautobot_content_type:
                raise ValueError(f"Can't specify both nautobot_content_type and extend_content_type {content_type}")
            extends_wrapper = self.wrappers[extend_content_type]
            nautobot_content_type = extends_wrapper.nautobot.content_type
        else:
            extends_wrapper = None

        if content_type in self.wrappers:
            wrapper = self.wrappers[content_type]
            if nautobot_content_type and wrapper.nautobot.content_type != nautobot_content_type:
                raise ValueError(
                    f"Content type {content_type} already mapped to {wrapper.nautobot.content_type} "
                    f"can't map to {nautobot_content_type}"
                )
        else:
            nautobot_wrapper = self.nautobot.get_or_create_wrapper(nautobot_content_type or content_type)
            wrapper = SourceModelWrapper(self, content_type, nautobot_wrapper)
            if not extends_wrapper:
                if nautobot_wrapper.content_type in self._content_types_back_mapping:
                    if self._content_types_back_mapping[nautobot_wrapper.content_type] != content_type:
                        self._content_types_back_mapping[nautobot_wrapper.content_type] = None
                else:
                    self._content_types_back_mapping[nautobot_wrapper.content_type] = content_type

        if extends_wrapper:
            wrapper.extends_wrapper = extends_wrapper

        if identifiers:
            wrapper.set_identifiers(identifiers)
        for field_name, definition in (fields or {}).items():
            wrapper.add_field(field_name, SourceFieldSource.CUSTOM).set_definition(definition)
        if default_reference:
            wrapper.set_default_reference(default_reference)
        if flags is not None:
            wrapper.flags = flags
        if nautobot_flags is not None:
            wrapper.nautobot.flags = nautobot_flags
        if pre_import:
            wrapper.pre_import = pre_import
        if disable_related_reference is not None:
            wrapper.disable_related_reference = disable_related_reference
        if forward_references:
            wrapper.forward_references = forward_references

        return wrapper

    def disable_model(self, content_type: ContentTypeStr, disable_reason: str) -> None:
        """Disable model importing."""
        self.get_or_create_wrapper(content_type).disable_reason = disable_reason

    def summarize(self, diffsync_summary: DiffSyncSummary) -> None:
        """Summarize the import."""
        self.summary.diffsync = diffsync_summary

        wrapper_to_id = {value: key for key, value in self.content_type_ids_mapping.items()}

        for content_type in sorted(self.wrappers):
            wrapper = self.wrappers.get(content_type)
            if wrapper:
                self.summary.source.append(wrapper.get_summary(wrapper_to_id.get(wrapper, None)))

        for content_type in sorted(self.nautobot.wrappers):
            wrapper = self.nautobot.wrappers.get(content_type)
            if wrapper:
                self.summary.nautobot.append(wrapper.get_summary())

    def get_or_create_wrapper(self, value: Union[None, SourceContentType]) -> "SourceModelWrapper":
        """Get a source Wrapper for a given content type."""
        # Enable mapping back from Nautobot content type, when using Nautobot model or wrapper
        map_back = False

        if not value:
            raise ValueError("Missing value")

        if isinstance(value, SourceModelWrapper):
            return value

        if isinstance(value, type(NautobotBaseModel)):
            map_back = True
            value = value._meta.label.lower()  # type: ignore
        elif isinstance(value, NautobotModelWrapper):
            map_back = True
            value = value.content_type

        if isinstance(value, str):
            value = value.lower()
        elif isinstance(value, int):
            if value not in self.content_type_ids_mapping:
                raise ValueError(f"Content type not found {value}")
            return self.content_type_ids_mapping[value]
        elif isinstance(value, Iterable) and len(value) == 2:
            value = ".".join(value).lower()
        else:
            raise ValueError(f"Invalid content type {value}")

        if map_back and value in self._content_types_back_mapping:
            back_mapping = self._content_types_back_mapping.get(value, None)
            if not back_mapping:
                raise ValueError(f"Ambiguous content type back mapping {value}")
            value = back_mapping

        if value in self.wrappers:
            return self.wrappers[value]

        return self.configure_model(value)

    def get_nautobot_content_type_uid(self, content_type: ContentTypeValue) -> int:
        """Get the Django content type ID for a given content type."""
        if isinstance(content_type, int):
            wrapper = self.content_type_ids_mapping.get(content_type, None)
            if not wrapper:
                raise ValueError(f"Content type not found {content_type}")
            return wrapper.nautobot.content_type_instance.pk
        if not isinstance(content_type, str):
            if not len(content_type) == 2:
                raise ValueError(f"Invalid content type {content_type}")
            content_type = ".".join(content_type)

        wrapper = self.get_or_create_wrapper(content_type)

        return wrapper.nautobot.content_type_instance.pk

    def load(self) -> None:
        """Load data from the source."""
        self.import_data()
        self.post_import()

    def import_data(self) -> None:
        """Import data from the source."""
        get_source_data = self.get_source_data

        # First pass to enhance pre-defined wrappers structure
        for content_type, data in get_source_data():
            if content_type in self.wrappers:
                wrapper = self.wrappers[content_type]
            else:
                wrapper = self.configure_model(content_type)
            wrapper.first_pass(data)

        # Create importers, wrappers structure is updated as needed
        while True:
            wrappers = [
                wrapper
                for wrapper in self.wrappers.values()
                if wrapper.importers is None and not wrapper.disable_reason
            ]
            if not wrappers:
                break
            for wrapper in wrappers:
                wrapper.create_importers()

        # Second pass to import actual data
        for content_type, data in get_source_data():
            self.wrappers[content_type].second_pass(data)

    def post_import(self) -> None:
        """Post import processing."""
        while any(wrapper.post_import() for wrapper in self.wrappers.values()):
            pass

        for nautobot_wrapper in self.get_imported_nautobot_wrappers():
            diffsync_class = nautobot_wrapper.diffsync_class
            # pylint: disable=protected-access
            model_name = diffsync_class._modelname
            self.top_level.append(model_name)
            setattr(self, model_name, diffsync_class)
            setattr(self.nautobot, model_name, getattr(self, model_name))

    def get_imported_nautobot_wrappers(self) -> Generator[NautobotModelWrapper, None, None]:
        """Get a list of Nautobot model wrappers in the order of import."""
        result = OrderedDict()

        for wrapper in self.wrappers.values():
            if (
                wrapper
                and not wrapper.disable_reason
                and wrapper.stats.created > 0
                and wrapper.nautobot.content_type not in result
            ):
                result[wrapper.nautobot.content_type] = wrapper.nautobot

        for content_type in IMPORT_ORDER:
            if content_type in result:
                yield result[content_type]
                del result[content_type]

        yield from result.values()
__init__(get_source_data, *args, nautobot=None, logger=None, **kwargs)

Initialize the SourceAdapter.

Source code in nautobot_netbox_importer/generator/source.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def __init__(
    self,
    get_source_data: SourceDataGenerator,
    *args,
    nautobot: Optional[NautobotAdapter] = None,
    logger=None,
    **kwargs,
):
    """Initialize the SourceAdapter."""
    super().__init__(*args, **kwargs)

    self.get_source_data = get_source_data
    self.wrappers: OrderedDict[ContentTypeStr, SourceModelWrapper] = OrderedDict()
    self.nautobot = nautobot or NautobotAdapter()
    self.content_type_ids_mapping: Dict[int, SourceModelWrapper] = {}
    self.logger = logger or default_logger
    self.summary = ImportSummary()

    # From Nautobot to Source content type mapping
    # When multiple source content types are mapped to the single nautobot content type, mapping is set to `None`
    self._content_types_back_mapping: Dict[ContentTypeStr, Optional[ContentTypeStr]] = {}
configure_model(content_type, nautobot_content_type='', extend_content_type='', identifiers=None, fields=None, default_reference=None, flags=None, nautobot_flags=None, pre_import=None, disable_related_reference=None, forward_references=None)

Create if not exist and configure a wrapper for a given source content type.

Create Nautobot content type wrapper as well.

Source code in nautobot_netbox_importer/generator/source.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def configure_model(
    self,
    content_type: ContentTypeStr,
    nautobot_content_type: ContentTypeStr = "",
    extend_content_type: ContentTypeStr = "",
    identifiers: Optional[Iterable[FieldName]] = None,
    fields: Optional[Mapping[FieldName, SourceFieldDefinition]] = None,
    default_reference: Optional[RecordData] = None,
    flags: Optional[DiffSyncModelFlags] = None,
    nautobot_flags: Optional[DiffSyncModelFlags] = None,
    pre_import: Optional[PreImport] = None,
    disable_related_reference: Optional[bool] = None,
    forward_references: Optional[ForwardReferences] = None,
) -> "SourceModelWrapper":
    """Create if not exist and configure a wrapper for a given source content type.

    Create Nautobot content type wrapper as well.
    """
    content_type = content_type.lower()
    nautobot_content_type = nautobot_content_type.lower()
    extend_content_type = extend_content_type.lower()

    if extend_content_type:
        if nautobot_content_type:
            raise ValueError(f"Can't specify both nautobot_content_type and extend_content_type {content_type}")
        extends_wrapper = self.wrappers[extend_content_type]
        nautobot_content_type = extends_wrapper.nautobot.content_type
    else:
        extends_wrapper = None

    if content_type in self.wrappers:
        wrapper = self.wrappers[content_type]
        if nautobot_content_type and wrapper.nautobot.content_type != nautobot_content_type:
            raise ValueError(
                f"Content type {content_type} already mapped to {wrapper.nautobot.content_type} "
                f"can't map to {nautobot_content_type}"
            )
    else:
        nautobot_wrapper = self.nautobot.get_or_create_wrapper(nautobot_content_type or content_type)
        wrapper = SourceModelWrapper(self, content_type, nautobot_wrapper)
        if not extends_wrapper:
            if nautobot_wrapper.content_type in self._content_types_back_mapping:
                if self._content_types_back_mapping[nautobot_wrapper.content_type] != content_type:
                    self._content_types_back_mapping[nautobot_wrapper.content_type] = None
            else:
                self._content_types_back_mapping[nautobot_wrapper.content_type] = content_type

    if extends_wrapper:
        wrapper.extends_wrapper = extends_wrapper

    if identifiers:
        wrapper.set_identifiers(identifiers)
    for field_name, definition in (fields or {}).items():
        wrapper.add_field(field_name, SourceFieldSource.CUSTOM).set_definition(definition)
    if default_reference:
        wrapper.set_default_reference(default_reference)
    if flags is not None:
        wrapper.flags = flags
    if nautobot_flags is not None:
        wrapper.nautobot.flags = nautobot_flags
    if pre_import:
        wrapper.pre_import = pre_import
    if disable_related_reference is not None:
        wrapper.disable_related_reference = disable_related_reference
    if forward_references:
        wrapper.forward_references = forward_references

    return wrapper
disable_model(content_type, disable_reason)

Disable model importing.

Source code in nautobot_netbox_importer/generator/source.py 
226
227
228
def disable_model(self, content_type: ContentTypeStr, disable_reason: str) -> None:
    """Disable model importing."""
    self.get_or_create_wrapper(content_type).disable_reason = disable_reason
get_imported_nautobot_wrappers()

Get a list of Nautobot model wrappers in the order of import.

Source code in nautobot_netbox_importer/generator/source.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
def get_imported_nautobot_wrappers(self) -> Generator[NautobotModelWrapper, None, None]:
    """Get a list of Nautobot model wrappers in the order of import."""
    result = OrderedDict()

    for wrapper in self.wrappers.values():
        if (
            wrapper
            and not wrapper.disable_reason
            and wrapper.stats.created > 0
            and wrapper.nautobot.content_type not in result
        ):
            result[wrapper.nautobot.content_type] = wrapper.nautobot

    for content_type in IMPORT_ORDER:
        if content_type in result:
            yield result[content_type]
            del result[content_type]

    yield from result.values()
get_nautobot_content_type_uid(content_type)

Get the Django content type ID for a given content type.

Source code in nautobot_netbox_importer/generator/source.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def get_nautobot_content_type_uid(self, content_type: ContentTypeValue) -> int:
    """Get the Django content type ID for a given content type."""
    if isinstance(content_type, int):
        wrapper = self.content_type_ids_mapping.get(content_type, None)
        if not wrapper:
            raise ValueError(f"Content type not found {content_type}")
        return wrapper.nautobot.content_type_instance.pk
    if not isinstance(content_type, str):
        if not len(content_type) == 2:
            raise ValueError(f"Invalid content type {content_type}")
        content_type = ".".join(content_type)

    wrapper = self.get_or_create_wrapper(content_type)

    return wrapper.nautobot.content_type_instance.pk
get_or_create_wrapper(value)

Get a source Wrapper for a given content type.

Source code in nautobot_netbox_importer/generator/source.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def get_or_create_wrapper(self, value: Union[None, SourceContentType]) -> "SourceModelWrapper":
    """Get a source Wrapper for a given content type."""
    # Enable mapping back from Nautobot content type, when using Nautobot model or wrapper
    map_back = False

    if not value:
        raise ValueError("Missing value")

    if isinstance(value, SourceModelWrapper):
        return value

    if isinstance(value, type(NautobotBaseModel)):
        map_back = True
        value = value._meta.label.lower()  # type: ignore
    elif isinstance(value, NautobotModelWrapper):
        map_back = True
        value = value.content_type

    if isinstance(value, str):
        value = value.lower()
    elif isinstance(value, int):
        if value not in self.content_type_ids_mapping:
            raise ValueError(f"Content type not found {value}")
        return self.content_type_ids_mapping[value]
    elif isinstance(value, Iterable) and len(value) == 2:
        value = ".".join(value).lower()
    else:
        raise ValueError(f"Invalid content type {value}")

    if map_back and value in self._content_types_back_mapping:
        back_mapping = self._content_types_back_mapping.get(value, None)
        if not back_mapping:
            raise ValueError(f"Ambiguous content type back mapping {value}")
        value = back_mapping

    if value in self.wrappers:
        return self.wrappers[value]

    return self.configure_model(value)
import_data()

Import data from the source.

Source code in nautobot_netbox_importer/generator/source.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
def import_data(self) -> None:
    """Import data from the source."""
    get_source_data = self.get_source_data

    # First pass to enhance pre-defined wrappers structure
    for content_type, data in get_source_data():
        if content_type in self.wrappers:
            wrapper = self.wrappers[content_type]
        else:
            wrapper = self.configure_model(content_type)
        wrapper.first_pass(data)

    # Create importers, wrappers structure is updated as needed
    while True:
        wrappers = [
            wrapper
            for wrapper in self.wrappers.values()
            if wrapper.importers is None and not wrapper.disable_reason
        ]
        if not wrappers:
            break
        for wrapper in wrappers:
            wrapper.create_importers()

    # Second pass to import actual data
    for content_type, data in get_source_data():
        self.wrappers[content_type].second_pass(data)
load()

Load data from the source.

Source code in nautobot_netbox_importer/generator/source.py 
302
303
304
305
def load(self) -> None:
    """Load data from the source."""
    self.import_data()
    self.post_import()
post_import()

Post import processing.

Source code in nautobot_netbox_importer/generator/source.py 
335
336
337
338
339
340
341
342
343
344
345
346
def post_import(self) -> None:
    """Post import processing."""
    while any(wrapper.post_import() for wrapper in self.wrappers.values()):
        pass

    for nautobot_wrapper in self.get_imported_nautobot_wrappers():
        diffsync_class = nautobot_wrapper.diffsync_class
        # pylint: disable=protected-access
        model_name = diffsync_class._modelname
        self.top_level.append(model_name)
        setattr(self, model_name, diffsync_class)
        setattr(self.nautobot, model_name, getattr(self, model_name))
summarize(diffsync_summary)

Summarize the import.

Source code in nautobot_netbox_importer/generator/source.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def summarize(self, diffsync_summary: DiffSyncSummary) -> None:
    """Summarize the import."""
    self.summary.diffsync = diffsync_summary

    wrapper_to_id = {value: key for key, value in self.content_type_ids_mapping.items()}

    for content_type in sorted(self.wrappers):
        wrapper = self.wrappers.get(content_type)
        if wrapper:
            self.summary.source.append(wrapper.get_summary(wrapper_to_id.get(wrapper, None)))

    for content_type in sorted(self.nautobot.wrappers):
        wrapper = self.nautobot.wrappers.get(content_type)
        if wrapper:
            self.summary.nautobot.append(wrapper.get_summary())

SourceField

Source Field.

Source code in nautobot_netbox_importer/generator/source.py 
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
class SourceField:
    """Source Field."""

    def __init__(self, wrapper: SourceModelWrapper, name: FieldName, source: SourceFieldSource):
        """Initialize the SourceField."""
        self.wrapper = wrapper
        wrapper.fields[name] = self
        self.name = name
        self.definition: SourceFieldDefinition = name
        self.sources = set((source,))
        self.processed = False
        self._nautobot: Optional[NautobotField] = None
        self.importer: Optional[SourceFieldImporter] = None
        self.default_value: Any = None
        self.disable_reason: str = ""

    def __str__(self) -> str:
        """Return a string representation of the field."""
        return self.wrapper.format_field_name(self.name)

    @property
    def nautobot(self) -> NautobotField:
        """Get the Nautobot field wrapper."""
        if not self._nautobot:
            raise RuntimeError(f"Missing Nautobot field for {self}")
        return self._nautobot

    def get_summary(self) -> FieldSummary:
        """Get a summary of the field."""
        return FieldSummary(
            name=self.name,
            nautobot_name=self._nautobot and self._nautobot.name,
            nautobot_internal_type=self._nautobot and self._nautobot.internal_type.value,
            nautobot_can_import=self._nautobot and self._nautobot.can_import,
            importer=self.importer and self.importer.__name__,
            definition=serialize_to_summary(self.definition),
            sources=sorted(source.name for source in self.sources),
            default_value=serialize_to_summary(self.default_value),
            disable_reason=self.disable_reason,
            required=self._nautobot.required if self._nautobot else False,
        )

    def disable(self, reason: str) -> None:
        """Disable field importing."""
        self.definition = None
        self.importer = None
        self.processed = True
        self.disable_reason = reason

    def handle_sibling(self, sibling: Union["SourceField", FieldName], nautobot_name: FieldName = "") -> "SourceField":
        """Specify, that this field importer handles other field."""
        if not self.importer:
            raise RuntimeError(f"Call `handle sibling` after setting importer for {self}")

        if isinstance(sibling, FieldName):
            sibling = self.wrapper.add_field(sibling, SourceFieldSource.SIBLING)

        sibling.set_nautobot_field(nautobot_name or self.nautobot.name)
        sibling.importer = self.importer
        sibling.processed = True

        if self.nautobot.can_import and not sibling.nautobot.can_import:
            self.disable(f"Can't import {self} based on {sibling}")

        return sibling

    def add_issue(self, issue_type: str, message: str, target: Optional[DiffSyncModel] = None) -> None:
        """Add an importer issue to the Nautobot Model Wrapper."""
        self.wrapper.nautobot.add_issue(issue_type, message, target=target, field=self.nautobot)

    def set_definition(self, definition: SourceFieldDefinition) -> None:
        """Customize field definition."""
        if self.processed:
            raise RuntimeError(f"Field already processed. {self}")

        if self.definition != definition:
            if self.definition != self.name:
                self.add_issue(
                    "OverrideDefinition",
                    f"Overriding field definition | Original: `{self.definition}` | New: `{definition}`",
                )
            self.definition = definition

    def create_importer(self) -> None:
        """Create importer for the field."""
        if self.processed:
            return
        self.processed = True

        if self.definition is None:
            return

        if isinstance(self.definition, FieldName):
            self.set_importer(nautobot_name=self.definition)
        elif callable(self.definition):
            self.definition(self)
        else:
            raise NotImplementedError(f"Unsupported field definition {self.definition}")

    def get_source_value(self, source: RecordData) -> Any:
        """Get a value from the source data, returning a default value if the value is empty."""
        if self.name not in source:
            return self.default_value

        result = source[self.name]
        return self.default_value if result in EMPTY_VALUES else result

    def set_nautobot_value(self, target: DiffSyncModel, value: Any) -> None:
        """Set a value to the Nautobot model."""
        if value in EMPTY_VALUES:
            if hasattr(target, self.nautobot.name):
                delattr(target, self.nautobot.name)
        else:
            setattr(target, self.nautobot.name, value)

    def set_nautobot_field(self, nautobot_name: FieldName = "") -> NautobotField:
        """Set a Nautobot field name for the field."""
        result = self.wrapper.nautobot.add_field(nautobot_name or self.name)
        if result.field:
            default_value = getattr(result.field, "default", None)
            if default_value not in EMPTY_VALUES and not isinstance(default_value, Callable):
                self.default_value = default_value
        self._nautobot = result
        if result.name == "last_updated":
            self.disable("Last updated field is updated with each write")
        return result

    # pylint: disable=too-many-branches
    def set_importer(
        self,
        importer: Optional[SourceFieldImporter] = None,
        nautobot_name: Optional[FieldName] = "",
        override=False,
    ) -> Optional[SourceFieldImporter]:
        """Sets the importer and Nautobot field if not already specified.

        If `nautobot_name` is not provided, the field name is used.

        Passing None to `nautobot_name` indicates that there is custom mapping without a direct relationship to a Nautobot field.
        """
        if self.disable_reason:
            raise RuntimeError(f"Can't set importer for disabled {self}")
        if self.importer and not override:
            raise RuntimeError(f"Importer already set for {self}")
        if not self._nautobot and nautobot_name is not None:
            self.set_nautobot_field(nautobot_name)

        if importer:
            self.importer = importer
            return importer

        if self.disable_reason or not self.nautobot.can_import:
            return None

        internal_type = self.nautobot.internal_type

        if internal_type == InternalFieldType.JSON_FIELD:
            self.set_json_importer()
        elif internal_type == InternalFieldType.DATE_FIELD:
            self.set_date_importer()
        elif internal_type == InternalFieldType.DATE_TIME_FIELD:
            self.set_datetime_importer()
        elif internal_type == InternalFieldType.UUID_FIELD:
            self.set_uuid_importer()
        elif internal_type == InternalFieldType.MANY_TO_MANY_FIELD:
            self.set_m2m_importer()
        elif internal_type == InternalFieldType.STATUS_FIELD:
            self.set_status_importer()
        elif self.nautobot.is_reference:
            self.set_relation_importer()
        elif getattr(self.nautobot.field, "choices", None):
            self.set_choice_importer()
        elif self.nautobot.is_integer:
            self.set_integer_importer()
        else:
            self.set_value_importer()

        return self.importer

    def set_value_importer(self) -> None:
        """Set a value importer."""

        def value_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            value = self.get_source_value(source)
            self.set_nautobot_value(target, value)

        self.set_importer(value_importer)

    def set_json_importer(self) -> None:
        """Set a JSON field importer."""

        def json_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            value = source.get(self.name, None)
            if isinstance(value, str) and value:
                value = json.loads(value)
            self.set_nautobot_value(target, value)

        self.set_importer(json_importer)

    def set_choice_importer(self) -> None:
        """Set a choice field importer."""
        field_choices = getattr(self.nautobot.field, "choices", None)
        if not field_choices:
            raise ValueError(f"Invalid field_choices for {self}")

        def get_choices(items: Iterable) -> Generator[Tuple[Any, Any], None, None]:
            for key, value in items:
                if isinstance(value, (list, tuple)):
                    yield from get_choices(value)
                else:
                    yield key, value

        choices = dict(get_choices(field_choices))

        def choice_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            value = self.get_source_value(source)
            if value in choices:
                self.set_nautobot_value(target, value)
            elif self.nautobot.required:
                # Set the choice value even it's not valid in Nautobot as it's required
                self.set_nautobot_value(target, value)
                raise InvalidChoiceValueIssue(self, value)
            elif value in EMPTY_VALUES:
                self.set_nautobot_value(target, value)
            else:
                raise InvalidChoiceValueIssue(self, value, None)

        self.set_importer(choice_importer)

    def set_integer_importer(self) -> None:
        """Set an integer field importer."""

        def integer_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            source_value = self.get_source_value(source)
            if source_value in EMPTY_VALUES:
                self.set_nautobot_value(target, source_value)
            else:
                source_value = float(source_value)
                value = int(source_value)
                self.set_nautobot_value(target, value)
                if value != source_value:
                    raise SourceFieldImporterIssue(f"Invalid source value {source_value}, truncated to {value}", self)

        self.set_importer(integer_importer)

    def set_relation_importer(self, related_wrapper: Optional[SourceModelWrapper] = None) -> None:
        """Set a relation importer."""
        wrapper = self.wrapper
        if not related_wrapper:
            if self.name == "parent":
                related_wrapper = wrapper
            else:
                related_wrapper = wrapper.adapter.get_or_create_wrapper(self.nautobot.related_model)

        if self.nautobot.is_content_type:
            self.set_content_type_importer()
            return

        if self.default_value in EMPTY_VALUES and related_wrapper.default_reference_uid:
            self.default_value = related_wrapper.default_reference_uid

        if not (self.default_value is None or isinstance(self.default_value, UUID)):
            raise NotImplementedError(f"Default value {self.default_value} is not a UUID")

        def relation_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            value = self.get_source_value(source)
            if value in EMPTY_VALUES:
                self.set_nautobot_value(target, value)
            else:
                if isinstance(value, (UUID, str, int)):
                    result = related_wrapper.get_pk_from_uid(value)
                else:
                    result = related_wrapper.get_pk_from_identifiers(value)
                self.set_nautobot_value(target, result)
                wrapper.add_reference(related_wrapper, result)

        self.set_importer(relation_importer)

    def set_content_type_importer(self) -> None:
        """Set a content type importer."""
        adapter = self.wrapper.adapter

        def content_type_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            content_type = source.get(self.name, None)
            if content_type not in EMPTY_VALUES:
                content_type = adapter.get_nautobot_content_type_uid(content_type)
            self.set_nautobot_value(target, content_type)

        self.set_importer(content_type_importer)

    def set_m2m_importer(self) -> None:
        """Set a many to many importer."""
        if not isinstance(self.nautobot.field, DjangoField):
            raise NotImplementedError(f"Unsupported m2m importer {self}")

        related_wrapper = self.wrapper.adapter.get_or_create_wrapper(self.nautobot.related_model)

        if related_wrapper.content_type == "contenttypes.contenttype":
            self.set_content_types_importer()
        elif related_wrapper.identifiers:
            self.set_identifiers_importer(related_wrapper)
        else:
            self.set_uuids_importer(related_wrapper)

    def set_identifiers_importer(self, related_wrapper: SourceModelWrapper) -> None:
        """Set a identifiers importer."""

        def identifiers_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            values = source.get(self.name, None)
            if values not in EMPTY_VALUES:
                if not isinstance(values, (list, set)):
                    raise ValueError(f"Invalid value {values} for field {self.name}")
                values = set(related_wrapper.get_pk_from_identifiers(item) for item in values)

            self.set_nautobot_value(target, values)

        self.set_importer(identifiers_importer)

    def set_content_types_importer(self) -> None:
        """Set a content types importer."""
        adapter = self.wrapper.adapter

        def content_types_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            values = source.get(self.name, None)
            if values in EMPTY_VALUES:
                return

            if not isinstance(values, (list, set)):
                raise ValueError(f"Invalid value {values} for field {self.name}")

            nautobot_values = set()
            for item in values:
                try:
                    nautobot_values.add(adapter.get_nautobot_content_type_uid(item))
                except NautobotModelNotFound:
                    self.add_issue("InvalidContentType", f"Invalid content type {item}, skipping", target)

            self.set_nautobot_value(target, nautobot_values)

        self.set_importer(content_types_importer)

    def set_uuids_importer(self, related_wrapper: SourceModelWrapper) -> None:
        """Set a UUIDs importer."""

        def uuids_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            value = source.get(self.name, None)
            if value in EMPTY_VALUES:
                self.set_nautobot_value(target, value)
            if isinstance(value, (UUID, str, int)):
                self.set_nautobot_value(target, {related_wrapper.get_pk_from_uid(value)})
            elif isinstance(value, (list, set, tuple)):
                self.set_nautobot_value(target, set(related_wrapper.get_pk_from_uid(item) for item in value))
            else:
                raise SourceFieldImporterIssue(f"Invalid value {value} for field {self.name}", self)

        self.set_importer(uuids_importer)

    def set_datetime_importer(self) -> None:
        """Set a datetime importer."""

        def datetime_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            value = source.get(self.name, None)
            if value not in EMPTY_VALUES:
                value = normalize_datetime(value)
            self.set_nautobot_value(target, value)

        self.set_importer(datetime_importer)

    def set_relation_and_type_importer(self, type_field: "SourceField") -> None:
        """Set a relation UUID importer based on the type field."""

        def relation_and_type_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            source_uid = source.get(self.name, None)
            source_type = source.get(type_field.name, None)
            if source_type in EMPTY_VALUES or source_uid in EMPTY_VALUES:
                if source_uid not in EMPTY_VALUES or source_type not in EMPTY_VALUES:
                    raise ValueError(
                        f"Both {self}=`{source_uid}` and {type_field}=`{source_type}` must be empty or not empty."
                    )
                return

            type_wrapper = self.wrapper.adapter.get_or_create_wrapper(source_type)
            uid = type_wrapper.get_pk_from_uid(source_uid)
            self.set_nautobot_value(target, uid)
            type_field.set_nautobot_value(target, type_wrapper.nautobot.content_type_instance.pk)
            self.wrapper.add_reference(type_wrapper, uid)

        self.set_importer(relation_and_type_importer)
        self.handle_sibling(type_field, type_field.name)

    def set_uuid_importer(self) -> None:
        """Set an UUID importer."""
        if self.name.endswith("_id"):
            type_field = self.wrapper.fields.get(self.name[:-3] + "_type", None)
            if type_field and type_field.nautobot.is_content_type:
                # Handles `<field name>_id` and `<field name>_type` fields combination
                self.set_relation_and_type_importer(type_field)
                return

        def uuid_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            value = source.get(self.name, None)
            if value not in EMPTY_VALUES:
                value = UUID(value)
            self.set_nautobot_value(target, value)

        self.set_importer(uuid_importer)

    def set_date_importer(self) -> None:
        """Set a date importer."""

        def date_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            value = source.get(self.name, None)
            if value not in EMPTY_VALUES and not isinstance(value, datetime.date):
                value = datetime.date.fromisoformat(str(value))
            self.set_nautobot_value(target, value)

        self.set_importer(date_importer)

    def set_status_importer(self) -> None:
        """Set a status importer."""
        status_wrapper = self.wrapper.adapter.get_or_create_wrapper("extras.status")
        if not self.default_value:
            self.default_value = status_wrapper.default_reference_uid

        if not (self.default_value is None or isinstance(self.default_value, UUID)):
            raise NotImplementedError(f"Default value {self.default_value} is not a UUID")

        def status_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
            status = source.get(self.name, None)
            if status:
                value = status_wrapper.cache_record({"name": status[0].upper() + status[1:]})
            else:
                value = self.default_value

            self.set_nautobot_value(target, value)
            if value:
                self.wrapper.add_reference(status_wrapper, value)

        self.set_importer(status_importer)
nautobot: NautobotField property

Get the Nautobot field wrapper.

__init__(wrapper, name, source)

Initialize the SourceField.

Source code in nautobot_netbox_importer/generator/source.py 
779
780
781
782
783
784
785
786
787
788
789
790
def __init__(self, wrapper: SourceModelWrapper, name: FieldName, source: SourceFieldSource):
    """Initialize the SourceField."""
    self.wrapper = wrapper
    wrapper.fields[name] = self
    self.name = name
    self.definition: SourceFieldDefinition = name
    self.sources = set((source,))
    self.processed = False
    self._nautobot: Optional[NautobotField] = None
    self.importer: Optional[SourceFieldImporter] = None
    self.default_value: Any = None
    self.disable_reason: str = ""
__str__()

Return a string representation of the field.

Source code in nautobot_netbox_importer/generator/source.py 
792
793
794
def __str__(self) -> str:
    """Return a string representation of the field."""
    return self.wrapper.format_field_name(self.name)
add_issue(issue_type, message, target=None)

Add an importer issue to the Nautobot Model Wrapper.

Source code in nautobot_netbox_importer/generator/source.py 
842
843
844
def add_issue(self, issue_type: str, message: str, target: Optional[DiffSyncModel] = None) -> None:
    """Add an importer issue to the Nautobot Model Wrapper."""
    self.wrapper.nautobot.add_issue(issue_type, message, target=target, field=self.nautobot)
create_importer()

Create importer for the field.

Source code in nautobot_netbox_importer/generator/source.py 
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
def create_importer(self) -> None:
    """Create importer for the field."""
    if self.processed:
        return
    self.processed = True

    if self.definition is None:
        return

    if isinstance(self.definition, FieldName):
        self.set_importer(nautobot_name=self.definition)
    elif callable(self.definition):
        self.definition(self)
    else:
        raise NotImplementedError(f"Unsupported field definition {self.definition}")
disable(reason)

Disable field importing.

Source code in nautobot_netbox_importer/generator/source.py 
818
819
820
821
822
823
def disable(self, reason: str) -> None:
    """Disable field importing."""
    self.definition = None
    self.importer = None
    self.processed = True
    self.disable_reason = reason
get_source_value(source)

Get a value from the source data, returning a default value if the value is empty.

Source code in nautobot_netbox_importer/generator/source.py 
875
876
877
878
879
880
881
def get_source_value(self, source: RecordData) -> Any:
    """Get a value from the source data, returning a default value if the value is empty."""
    if self.name not in source:
        return self.default_value

    result = source[self.name]
    return self.default_value if result in EMPTY_VALUES else result
get_summary()

Get a summary of the field.

Source code in nautobot_netbox_importer/generator/source.py 
803
804
805
806
807
808
809
810
811
812
813
814
815
816
def get_summary(self) -> FieldSummary:
    """Get a summary of the field."""
    return FieldSummary(
        name=self.name,
        nautobot_name=self._nautobot and self._nautobot.name,
        nautobot_internal_type=self._nautobot and self._nautobot.internal_type.value,
        nautobot_can_import=self._nautobot and self._nautobot.can_import,
        importer=self.importer and self.importer.__name__,
        definition=serialize_to_summary(self.definition),
        sources=sorted(source.name for source in self.sources),
        default_value=serialize_to_summary(self.default_value),
        disable_reason=self.disable_reason,
        required=self._nautobot.required if self._nautobot else False,
    )
handle_sibling(sibling, nautobot_name='')

Specify, that this field importer handles other field.

Source code in nautobot_netbox_importer/generator/source.py 
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
def handle_sibling(self, sibling: Union["SourceField", FieldName], nautobot_name: FieldName = "") -> "SourceField":
    """Specify, that this field importer handles other field."""
    if not self.importer:
        raise RuntimeError(f"Call `handle sibling` after setting importer for {self}")

    if isinstance(sibling, FieldName):
        sibling = self.wrapper.add_field(sibling, SourceFieldSource.SIBLING)

    sibling.set_nautobot_field(nautobot_name or self.nautobot.name)
    sibling.importer = self.importer
    sibling.processed = True

    if self.nautobot.can_import and not sibling.nautobot.can_import:
        self.disable(f"Can't import {self} based on {sibling}")

    return sibling
set_choice_importer()

Set a choice field importer.

Source code in nautobot_netbox_importer/generator/source.py 
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
def set_choice_importer(self) -> None:
    """Set a choice field importer."""
    field_choices = getattr(self.nautobot.field, "choices", None)
    if not field_choices:
        raise ValueError(f"Invalid field_choices for {self}")

    def get_choices(items: Iterable) -> Generator[Tuple[Any, Any], None, None]:
        for key, value in items:
            if isinstance(value, (list, tuple)):
                yield from get_choices(value)
            else:
                yield key, value

    choices = dict(get_choices(field_choices))

    def choice_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        value = self.get_source_value(source)
        if value in choices:
            self.set_nautobot_value(target, value)
        elif self.nautobot.required:
            # Set the choice value even it's not valid in Nautobot as it's required
            self.set_nautobot_value(target, value)
            raise InvalidChoiceValueIssue(self, value)
        elif value in EMPTY_VALUES:
            self.set_nautobot_value(target, value)
        else:
            raise InvalidChoiceValueIssue(self, value, None)

    self.set_importer(choice_importer)
set_content_type_importer()

Set a content type importer.

Source code in nautobot_netbox_importer/generator/source.py 
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
def set_content_type_importer(self) -> None:
    """Set a content type importer."""
    adapter = self.wrapper.adapter

    def content_type_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        content_type = source.get(self.name, None)
        if content_type not in EMPTY_VALUES:
            content_type = adapter.get_nautobot_content_type_uid(content_type)
        self.set_nautobot_value(target, content_type)

    self.set_importer(content_type_importer)
set_content_types_importer()

Set a content types importer.

Source code in nautobot_netbox_importer/generator/source.py 
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
def set_content_types_importer(self) -> None:
    """Set a content types importer."""
    adapter = self.wrapper.adapter

    def content_types_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        values = source.get(self.name, None)
        if values in EMPTY_VALUES:
            return

        if not isinstance(values, (list, set)):
            raise ValueError(f"Invalid value {values} for field {self.name}")

        nautobot_values = set()
        for item in values:
            try:
                nautobot_values.add(adapter.get_nautobot_content_type_uid(item))
            except NautobotModelNotFound:
                self.add_issue("InvalidContentType", f"Invalid content type {item}, skipping", target)

        self.set_nautobot_value(target, nautobot_values)

    self.set_importer(content_types_importer)
set_date_importer()

Set a date importer.

Source code in nautobot_netbox_importer/generator/source.py 
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
def set_date_importer(self) -> None:
    """Set a date importer."""

    def date_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        value = source.get(self.name, None)
        if value not in EMPTY_VALUES and not isinstance(value, datetime.date):
            value = datetime.date.fromisoformat(str(value))
        self.set_nautobot_value(target, value)

    self.set_importer(date_importer)
set_datetime_importer()

Set a datetime importer.

Source code in nautobot_netbox_importer/generator/source.py 
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
def set_datetime_importer(self) -> None:
    """Set a datetime importer."""

    def datetime_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        value = source.get(self.name, None)
        if value not in EMPTY_VALUES:
            value = normalize_datetime(value)
        self.set_nautobot_value(target, value)

    self.set_importer(datetime_importer)
set_definition(definition)

Customize field definition.

Source code in nautobot_netbox_importer/generator/source.py 
846
847
848
849
850
851
852
853
854
855
856
857
def set_definition(self, definition: SourceFieldDefinition) -> None:
    """Customize field definition."""
    if self.processed:
        raise RuntimeError(f"Field already processed. {self}")

    if self.definition != definition:
        if self.definition != self.name:
            self.add_issue(
                "OverrideDefinition",
                f"Overriding field definition | Original: `{self.definition}` | New: `{definition}`",
            )
        self.definition = definition
set_identifiers_importer(related_wrapper)

Set a identifiers importer.

Source code in nautobot_netbox_importer/generator/source.py 
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
def set_identifiers_importer(self, related_wrapper: SourceModelWrapper) -> None:
    """Set a identifiers importer."""

    def identifiers_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        values = source.get(self.name, None)
        if values not in EMPTY_VALUES:
            if not isinstance(values, (list, set)):
                raise ValueError(f"Invalid value {values} for field {self.name}")
            values = set(related_wrapper.get_pk_from_identifiers(item) for item in values)

        self.set_nautobot_value(target, values)

    self.set_importer(identifiers_importer)
set_importer(importer=None, nautobot_name='', override=False)

Sets the importer and Nautobot field if not already specified.

If nautobot_name is not provided, the field name is used.

Passing None to nautobot_name indicates that there is custom mapping without a direct relationship to a Nautobot field.

Source code in nautobot_netbox_importer/generator/source.py 
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
def set_importer(
    self,
    importer: Optional[SourceFieldImporter] = None,
    nautobot_name: Optional[FieldName] = "",
    override=False,
) -> Optional[SourceFieldImporter]:
    """Sets the importer and Nautobot field if not already specified.

    If `nautobot_name` is not provided, the field name is used.

    Passing None to `nautobot_name` indicates that there is custom mapping without a direct relationship to a Nautobot field.
    """
    if self.disable_reason:
        raise RuntimeError(f"Can't set importer for disabled {self}")
    if self.importer and not override:
        raise RuntimeError(f"Importer already set for {self}")
    if not self._nautobot and nautobot_name is not None:
        self.set_nautobot_field(nautobot_name)

    if importer:
        self.importer = importer
        return importer

    if self.disable_reason or not self.nautobot.can_import:
        return None

    internal_type = self.nautobot.internal_type

    if internal_type == InternalFieldType.JSON_FIELD:
        self.set_json_importer()
    elif internal_type == InternalFieldType.DATE_FIELD:
        self.set_date_importer()
    elif internal_type == InternalFieldType.DATE_TIME_FIELD:
        self.set_datetime_importer()
    elif internal_type == InternalFieldType.UUID_FIELD:
        self.set_uuid_importer()
    elif internal_type == InternalFieldType.MANY_TO_MANY_FIELD:
        self.set_m2m_importer()
    elif internal_type == InternalFieldType.STATUS_FIELD:
        self.set_status_importer()
    elif self.nautobot.is_reference:
        self.set_relation_importer()
    elif getattr(self.nautobot.field, "choices", None):
        self.set_choice_importer()
    elif self.nautobot.is_integer:
        self.set_integer_importer()
    else:
        self.set_value_importer()

    return self.importer
set_integer_importer()

Set an integer field importer.

Source code in nautobot_netbox_importer/generator/source.py 
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
def set_integer_importer(self) -> None:
    """Set an integer field importer."""

    def integer_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        source_value = self.get_source_value(source)
        if source_value in EMPTY_VALUES:
            self.set_nautobot_value(target, source_value)
        else:
            source_value = float(source_value)
            value = int(source_value)
            self.set_nautobot_value(target, value)
            if value != source_value:
                raise SourceFieldImporterIssue(f"Invalid source value {source_value}, truncated to {value}", self)

    self.set_importer(integer_importer)
set_json_importer()

Set a JSON field importer.

Source code in nautobot_netbox_importer/generator/source.py 
964
965
966
967
968
969
970
971
972
973
def set_json_importer(self) -> None:
    """Set a JSON field importer."""

    def json_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        value = source.get(self.name, None)
        if isinstance(value, str) and value:
            value = json.loads(value)
        self.set_nautobot_value(target, value)

    self.set_importer(json_importer)
set_m2m_importer()

Set a many to many importer.

Source code in nautobot_netbox_importer/generator/source.py 
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
def set_m2m_importer(self) -> None:
    """Set a many to many importer."""
    if not isinstance(self.nautobot.field, DjangoField):
        raise NotImplementedError(f"Unsupported m2m importer {self}")

    related_wrapper = self.wrapper.adapter.get_or_create_wrapper(self.nautobot.related_model)

    if related_wrapper.content_type == "contenttypes.contenttype":
        self.set_content_types_importer()
    elif related_wrapper.identifiers:
        self.set_identifiers_importer(related_wrapper)
    else:
        self.set_uuids_importer(related_wrapper)
set_nautobot_field(nautobot_name='')

Set a Nautobot field name for the field.

Source code in nautobot_netbox_importer/generator/source.py 
891
892
893
894
895
896
897
898
899
900
901
def set_nautobot_field(self, nautobot_name: FieldName = "") -> NautobotField:
    """Set a Nautobot field name for the field."""
    result = self.wrapper.nautobot.add_field(nautobot_name or self.name)
    if result.field:
        default_value = getattr(result.field, "default", None)
        if default_value not in EMPTY_VALUES and not isinstance(default_value, Callable):
            self.default_value = default_value
    self._nautobot = result
    if result.name == "last_updated":
        self.disable("Last updated field is updated with each write")
    return result
set_nautobot_value(target, value)

Set a value to the Nautobot model.

Source code in nautobot_netbox_importer/generator/source.py 
883
884
885
886
887
888
889
def set_nautobot_value(self, target: DiffSyncModel, value: Any) -> None:
    """Set a value to the Nautobot model."""
    if value in EMPTY_VALUES:
        if hasattr(target, self.nautobot.name):
            delattr(target, self.nautobot.name)
    else:
        setattr(target, self.nautobot.name, value)
set_relation_and_type_importer(type_field)

Set a relation UUID importer based on the type field.

Source code in nautobot_netbox_importer/generator/source.py 
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
def set_relation_and_type_importer(self, type_field: "SourceField") -> None:
    """Set a relation UUID importer based on the type field."""

    def relation_and_type_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        source_uid = source.get(self.name, None)
        source_type = source.get(type_field.name, None)
        if source_type in EMPTY_VALUES or source_uid in EMPTY_VALUES:
            if source_uid not in EMPTY_VALUES or source_type not in EMPTY_VALUES:
                raise ValueError(
                    f"Both {self}=`{source_uid}` and {type_field}=`{source_type}` must be empty or not empty."
                )
            return

        type_wrapper = self.wrapper.adapter.get_or_create_wrapper(source_type)
        uid = type_wrapper.get_pk_from_uid(source_uid)
        self.set_nautobot_value(target, uid)
        type_field.set_nautobot_value(target, type_wrapper.nautobot.content_type_instance.pk)
        self.wrapper.add_reference(type_wrapper, uid)

    self.set_importer(relation_and_type_importer)
    self.handle_sibling(type_field, type_field.name)
set_relation_importer(related_wrapper=None)

Set a relation importer.

Source code in nautobot_netbox_importer/generator/source.py 
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
def set_relation_importer(self, related_wrapper: Optional[SourceModelWrapper] = None) -> None:
    """Set a relation importer."""
    wrapper = self.wrapper
    if not related_wrapper:
        if self.name == "parent":
            related_wrapper = wrapper
        else:
            related_wrapper = wrapper.adapter.get_or_create_wrapper(self.nautobot.related_model)

    if self.nautobot.is_content_type:
        self.set_content_type_importer()
        return

    if self.default_value in EMPTY_VALUES and related_wrapper.default_reference_uid:
        self.default_value = related_wrapper.default_reference_uid

    if not (self.default_value is None or isinstance(self.default_value, UUID)):
        raise NotImplementedError(f"Default value {self.default_value} is not a UUID")

    def relation_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        value = self.get_source_value(source)
        if value in EMPTY_VALUES:
            self.set_nautobot_value(target, value)
        else:
            if isinstance(value, (UUID, str, int)):
                result = related_wrapper.get_pk_from_uid(value)
            else:
                result = related_wrapper.get_pk_from_identifiers(value)
            self.set_nautobot_value(target, result)
            wrapper.add_reference(related_wrapper, result)

    self.set_importer(relation_importer)
set_status_importer()

Set a status importer.

Source code in nautobot_netbox_importer/generator/source.py 
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
def set_status_importer(self) -> None:
    """Set a status importer."""
    status_wrapper = self.wrapper.adapter.get_or_create_wrapper("extras.status")
    if not self.default_value:
        self.default_value = status_wrapper.default_reference_uid

    if not (self.default_value is None or isinstance(self.default_value, UUID)):
        raise NotImplementedError(f"Default value {self.default_value} is not a UUID")

    def status_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        status = source.get(self.name, None)
        if status:
            value = status_wrapper.cache_record({"name": status[0].upper() + status[1:]})
        else:
            value = self.default_value

        self.set_nautobot_value(target, value)
        if value:
            self.wrapper.add_reference(status_wrapper, value)

    self.set_importer(status_importer)
set_uuid_importer()

Set an UUID importer.

Source code in nautobot_netbox_importer/generator/source.py 
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
def set_uuid_importer(self) -> None:
    """Set an UUID importer."""
    if self.name.endswith("_id"):
        type_field = self.wrapper.fields.get(self.name[:-3] + "_type", None)
        if type_field and type_field.nautobot.is_content_type:
            # Handles `<field name>_id` and `<field name>_type` fields combination
            self.set_relation_and_type_importer(type_field)
            return

    def uuid_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        value = source.get(self.name, None)
        if value not in EMPTY_VALUES:
            value = UUID(value)
        self.set_nautobot_value(target, value)

    self.set_importer(uuid_importer)
set_uuids_importer(related_wrapper)

Set a UUIDs importer.

Source code in nautobot_netbox_importer/generator/source.py 
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
def set_uuids_importer(self, related_wrapper: SourceModelWrapper) -> None:
    """Set a UUIDs importer."""

    def uuids_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        value = source.get(self.name, None)
        if value in EMPTY_VALUES:
            self.set_nautobot_value(target, value)
        if isinstance(value, (UUID, str, int)):
            self.set_nautobot_value(target, {related_wrapper.get_pk_from_uid(value)})
        elif isinstance(value, (list, set, tuple)):
            self.set_nautobot_value(target, set(related_wrapper.get_pk_from_uid(item) for item in value))
        else:
            raise SourceFieldImporterIssue(f"Invalid value {value} for field {self.name}", self)

    self.set_importer(uuids_importer)
set_value_importer()

Set a value importer.

Source code in nautobot_netbox_importer/generator/source.py 
955
956
957
958
959
960
961
962
def set_value_importer(self) -> None:
    """Set a value importer."""

    def value_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
        value = self.get_source_value(source)
        self.set_nautobot_value(target, value)

    self.set_importer(value_importer)

SourceFieldImporterIssue

Bases: NetBoxImporterException

Raised when an error occurs during field import.

Raising this exception gathers the issue and continues with the next importer.

Source code in nautobot_netbox_importer/generator/source.py 
59
60
61
62
63
64
65
66
67
68
class SourceFieldImporterIssue(NetBoxImporterException):
    """Raised when an error occurs during field import.

    Raising this exception gathers the issue and continues with the next importer.
    """

    def __init__(self, message: str, field: "SourceField"):
        """Initialize the exception."""
        super().__init__(message)
        self.field = field
__init__(message, field)

Initialize the exception.

Source code in nautobot_netbox_importer/generator/source.py 
65
66
67
68
def __init__(self, message: str, field: "SourceField"):
    """Initialize the exception."""
    super().__init__(message)
    self.field = field

SourceModelWrapper

Definition of a source model mapping to Nautobot model.

Source code in nautobot_netbox_importer/generator/source.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
class SourceModelWrapper:
    """Definition of a source model mapping to Nautobot model."""

    def __init__(self, adapter: SourceAdapter, content_type: ContentTypeStr, nautobot_wrapper: NautobotModelWrapper):
        """Initialize the SourceModelWrapper."""
        if content_type in adapter.wrappers:
            raise ValueError(f"Duplicate content type {content_type}")
        adapter.wrappers[content_type] = self
        self.adapter = adapter
        self.content_type = content_type
        self.nautobot = nautobot_wrapper
        if self.nautobot.disabled:
            self.disable_reason = f"Nautobot content type: `{nautobot_wrapper.content_type}` not found"
        else:
            self.disable_reason = ""

        # Source field names when referencing this model
        self.identifiers: Optional[List[FieldName]] = None

        # Used to autofill `content_types` field
        self.disable_related_reference = False
        self.references: SourceReferences = {}
        self.forward_references: Optional[ForwardReferences] = None

        # Whether importing record data exteds existing record
        self.extends_wrapper: Optional[SourceModelWrapper] = None

        # Importers are created after all fields are defined
        self.importers: Optional[Set[SourceFieldImporter]] = None

        # Default reference to this model
        self.default_reference_uid: Optional[Uid] = None

        # Caching
        self._uid_to_pk_cache: Dict[Uid, Uid] = {}
        self._cached_data: Dict[Uid, RecordData] = {}

        self.stats = SourceModelStats()
        self.flags = DiffSyncModelFlags.NONE

        # Source fields defintions
        self.fields: OrderedDict[FieldName, SourceField] = OrderedDict()
        self.pre_import: Optional[PreImport] = None

        if self.disable_reason:
            self.adapter.logger.debug("Created disabled %s", self)
            return

        pk_field = self.add_field(nautobot_wrapper.pk_field.name, SourceFieldSource.AUTO)
        pk_field.set_nautobot_field()
        pk_field.processed = True

        if issubclass(nautobot_wrapper.model, TreeModel):
            for name in ("tree_id", "lft", "rght", "level"):
                self.disable_field(name, "Tree fields doesn't need to be imported")

        self.adapter.logger.debug("Created %s", self)

    def __str__(self) -> str:
        """Return a string representation of the wrapper."""
        return f"{self.__class__.__name__}<{self.content_type} -> {self.nautobot.content_type}>"

    def cache_record_uids(self, source: RecordData, nautobot_uid: Optional[Uid] = None) -> Uid:
        """Cache record identifier mappings.

        When `nautobot_uid` is not provided, it is generated from the source data and caching is processed there.
        """
        if not nautobot_uid:
            return self.get_pk_from_data(source)

        if self.identifiers:
            identifiers_data = [source[field_name] for field_name in self.identifiers]
            self._uid_to_pk_cache[json.dumps(identifiers_data)] = nautobot_uid

        source_uid = source.get(self.nautobot.pk_field.name, None)
        if source_uid and source_uid not in self._uid_to_pk_cache:
            self._uid_to_pk_cache[source_uid] = nautobot_uid

        self._uid_to_pk_cache[nautobot_uid] = nautobot_uid

        return nautobot_uid

    def first_pass(self, data: RecordData) -> None:
        """Firts pass of data import."""
        if self.pre_import:
            if self.pre_import(data, ImporterPass.DEFINE_STRUCTURE) != PreImportResult.USE_RECORD:
                self.stats.first_pass_skipped += 1
                return

        self.stats.first_pass_used += 1

        if self.disable_reason:
            return

        for field_name in data.keys():
            self.add_field(field_name, SourceFieldSource.DATA)

    def second_pass(self, data: RecordData) -> None:
        """Second pass of data import."""
        if self.disable_reason:
            return

        if self.pre_import:
            if self.pre_import(data, ImporterPass.IMPORT_DATA) != PreImportResult.USE_RECORD:
                self.stats.second_pass_skipped += 1
                return

        self.stats.second_pass_used += 1

        self.import_record(data)

    def get_summary(self, content_type_id) -> SourceModelSummary:
        """Get a summary of the model."""
        fields = [field.get_summary() for field in self.fields.values()]

        return SourceModelSummary(
            content_type=self.content_type,
            content_type_id=content_type_id,
            extends_content_type=self.extends_wrapper and self.extends_wrapper.content_type,
            nautobot_content_type=self.nautobot.content_type,
            disable_reason=self.disable_reason,
            identifiers=self.identifiers,
            disable_related_reference=self.disable_related_reference,
            forward_references=self.forward_references and self.forward_references.__name__ or None,
            pre_import=self.pre_import and self.pre_import.__name__ or None,
            fields=sorted(fields, key=lambda field: field.name),
            flags=str(self.flags),
            default_reference_uid=serialize_to_summary(self.default_reference_uid),
            stats=self.stats,
        )

    def set_identifiers(self, identifiers: Iterable[FieldName]) -> None:
        """Set identifiers for the model."""
        if self.identifiers:
            if list(identifiers) == self.identifiers:
                return
            raise ValueError(
                f"Different identifiers were already set up | original: `{self.identifiers}` | new: `{identifiers}`"
            )

        if list(identifiers) == [self.nautobot.pk_field.name]:
            return

        self.identifiers = list(identifiers)
        for identifier in self.identifiers:
            self.add_field(identifier, SourceFieldSource.IDENTIFIER)

    def disable_field(self, field_name: FieldName, reason: str) -> "SourceField":
        """Disable field importing."""
        field = self.add_field(field_name, SourceFieldSource.CUSTOM)
        field.disable(reason)
        return field

    def format_field_name(self, name: FieldName) -> str:
        """Format a field name for logging."""
        return f"{self.content_type}->{name}"

    def add_field(self, name: FieldName, source: SourceFieldSource) -> "SourceField":
        """Add a field definition for a source field."""
        if self.importers is not None:
            raise ValueError(f"Can't add field {self.format_field_name(name)}, model's importers already created.")

        if name not in self.fields:
            return SourceField(self, name, source)

        field = self.fields[name]
        field.sources.add(source)
        return field

    def create_importers(self) -> None:
        """Create importers for all fields."""
        if self.importers is not None:
            raise RuntimeError(f"Importers already created for {self.content_type}")

        if not self.extends_wrapper:
            for field_name in AUTO_ADD_FIELDS:
                if hasattr(self.nautobot.model, field_name):
                    self.add_field(field_name, SourceFieldSource.AUTO)

        while True:
            fields = [field for field in self.fields.values() if not field.processed]
            if not fields:
                break

            for field in fields:
                try:
                    field.create_importer()
                except Exception:
                    self.adapter.logger.error("Failed to create importer for %s", field)
                    raise

        self.importers = set(field.importer for field in self.fields.values() if field.importer)

    def get_pk_from_uid(self, uid: Uid) -> Uid:
        """Get a source primary key for a given source uid."""
        if uid in self._uid_to_pk_cache:
            return self._uid_to_pk_cache[uid]

        if self.nautobot.pk_field.internal_type == InternalFieldType.UUID_FIELD:
            if self.extends_wrapper:
                result = self.extends_wrapper.get_pk_from_uid(uid)
            else:
                result = source_pk_to_uuid(self.content_type or self.content_type, uid)
        elif self.nautobot.pk_field.is_auto_increment:
            self.nautobot.last_id += 1
            result = self.nautobot.last_id
        else:
            raise ValueError(f"Unsupported pk_type {self.nautobot.pk_field.internal_type}")

        self._uid_to_pk_cache[uid] = result
        self._uid_to_pk_cache[result] = result

        return result

    def get_pk_from_identifiers(self, data: Union[Uid, Iterable[Uid]]) -> Uid:
        """Get a source primary key for a given source identifiers."""
        if not self.identifiers:
            if isinstance(data, (UUID, str, int)):
                return self.get_pk_from_uid(data)

            raise ValueError(f"Invalid identifiers {data} for {self.identifiers}")

        if not isinstance(data, list):
            data = list(data)  # type: ignore
        if len(self.identifiers) != len(data):
            raise ValueError(f"Invalid identifiers {data} for {self.identifiers}")

        identifiers_uid = json.dumps(data)
        if identifiers_uid in self._uid_to_pk_cache:
            return self._uid_to_pk_cache[identifiers_uid]

        filter_kwargs = {self.identifiers[index]: value for index, value in enumerate(data)}
        try:
            nautobot_instance = self.nautobot.model.objects.get(**filter_kwargs)
            nautobot_uid = getattr(nautobot_instance, self.nautobot.pk_field.name)
            if not nautobot_uid:
                raise ValueError(f"Invalid args {filter_kwargs} for {nautobot_instance}")
            self._uid_to_pk_cache[identifiers_uid] = nautobot_uid
            self._uid_to_pk_cache[nautobot_uid] = nautobot_uid
            return nautobot_uid
        except self.nautobot.model.DoesNotExist:  # type: ignore
            return self.get_pk_from_uid(identifiers_uid)

    def get_pk_from_data(self, data: RecordData) -> Uid:
        """Get a source primary key for a given source data."""
        if not self.identifiers:
            return self.get_pk_from_uid(data[self.nautobot.pk_field.name])

        data_uid = data.get(self.nautobot.pk_field.name, None)
        if data_uid and data_uid in self._uid_to_pk_cache:
            return self._uid_to_pk_cache[data_uid]

        result = self.get_pk_from_identifiers(data[field_name] for field_name in self.identifiers)

        if data_uid:
            self._uid_to_pk_cache[data_uid] = result

        return result

    def import_record(self, data: RecordData, target: Optional[DiffSyncBaseModel] = None) -> DiffSyncBaseModel:
        """Import a single item from the source."""
        self.adapter.logger.debug("Importing record %s %s", self, data)
        if self.importers is None:
            raise RuntimeError(f"Importers not created for {self}")

        if target:
            uid = getattr(target, self.nautobot.pk_field.name)
        else:
            uid = self.get_pk_from_data(data)
            target = self.get_or_create(uid)

        for importer in self.importers:
            try:
                importer(data, target)
            # pylint: disable=broad-exception-caught
            except Exception as error:
                self.nautobot.add_issue(
                    error.__class__.__name__,
                    getattr(error, "message", "") or str(error),
                    target=target,
                    field=error.field.nautobot if isinstance(error, SourceFieldImporterIssue) else None,
                )

        self.stats.imported += 1
        self.adapter.logger.debug("Imported %s %s", uid, target.get_attrs())

        return target

    def get_or_create(self, uid: Uid, fail_missing=False) -> DiffSyncBaseModel:
        """Get an existing DiffSync Model instance from the source or create a new one.

        Use Nautobot data as defaults if available.
        """
        filter_kwargs = {self.nautobot.pk_field.name: uid}
        diffsync_class = self.nautobot.diffsync_class
        result = self.adapter.get_or_none(diffsync_class, filter_kwargs)
        if result:
            if not isinstance(result, DiffSyncBaseModel):
                raise TypeError(f"Invalid instance type {result}")
            return result

        result = diffsync_class(**filter_kwargs, diffsync=self.adapter)  # type: ignore
        result.model_flags = self.flags

        cached_data = self._cached_data.get(uid, None)
        if cached_data:
            fail_missing = False
            self.import_record(cached_data, result)
            self.stats.imported_from_cache += 1

        nautobot_diffsync_instance = self.nautobot.find_or_create(filter_kwargs)
        if nautobot_diffsync_instance:
            fail_missing = False
            for key, value in nautobot_diffsync_instance.get_attrs().items():
                if value not in EMPTY_VALUES:
                    setattr(result, key, value)

        if fail_missing:
            raise ValueError(f"Missing {self} {uid} in Nautobot or cached data")

        self.adapter.add(result)
        self.stats.created += 1
        if self.flags == DiffSyncModelFlags.IGNORE:
            self.nautobot.stats.source_ignored += 1
        else:
            self.nautobot.stats.source_created += 1

        return result

    def get_default_reference_uid(self) -> Uid:
        """Get the default reference to this model."""
        if self.default_reference_uid:
            return self.default_reference_uid
        raise ValueError("Missing default reference")

    def cache_record(self, data: RecordData) -> Uid:
        """Cache data for optional later use.

        If record is referenced by other models, it will be imported automatically; otherwise, it will be ignored.
        """
        uid = self.get_pk_from_data(data)
        if uid in self._cached_data:
            return uid

        if self.importers is None:
            for field_name in data.keys():
                self.add_field(field_name, SourceFieldSource.CACHE)

        self._cached_data[uid] = data
        self.stats.pre_cached += 1

        self.adapter.logger.debug("Cached %s %s %s", self, uid, data)

        return uid

    def set_default_reference(self, data: RecordData) -> None:
        """Set the default reference to this model."""
        self.default_reference_uid = self.cache_record(data)

    def post_import(self) -> bool:
        """Post import processing.

        Assigns referenced content_types to the imported models.

        Returns False if no post processing is needed, otherwise True.
        """
        if not self.references:
            return False

        references = self.references
        self.references = {}

        if self.forward_references:
            self.forward_references(self, references)
            return True

        for uid, content_types in references.items():
            # Keep this even when no content_types field is present, to create referenced cached data
            instance = self.get_or_create(uid, fail_missing=True)
            if "content_types" not in self.nautobot.fields:
                continue

            content_types = set(wrapper.nautobot.content_type_instance.pk for wrapper in content_types)
            target_content_types = getattr(instance, "content_types", None)
            if target_content_types != content_types:
                if target_content_types:
                    target_content_types.update(content_types)
                else:
                    instance.content_types = content_types
                self.adapter.update(instance)

        return True

    def add_reference(self, related_wrapper: "SourceModelWrapper", uid: Uid) -> None:
        """Add a reference from this content type to related record."""
        if self.disable_related_reference:
            return
        self.adapter.logger.debug(
            "Adding reference from: %s to: %s %s", self.content_type, related_wrapper.content_type, uid
        )
        if not uid:
            raise ValueError(f"Invalid uid {uid}")
        related_wrapper.references.setdefault(uid, set()).add(self)
__init__(adapter, content_type, nautobot_wrapper)

Initialize the SourceModelWrapper.

Source code in nautobot_netbox_importer/generator/source.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
def __init__(self, adapter: SourceAdapter, content_type: ContentTypeStr, nautobot_wrapper: NautobotModelWrapper):
    """Initialize the SourceModelWrapper."""
    if content_type in adapter.wrappers:
        raise ValueError(f"Duplicate content type {content_type}")
    adapter.wrappers[content_type] = self
    self.adapter = adapter
    self.content_type = content_type
    self.nautobot = nautobot_wrapper
    if self.nautobot.disabled:
        self.disable_reason = f"Nautobot content type: `{nautobot_wrapper.content_type}` not found"
    else:
        self.disable_reason = ""