Building a Nautobot SSoT App

In a previous post we established the importance of having a single source of truth (SSoT), provided an overview of the Nautobot Single Source of Truth (SSoT) framework, and how the SSoT framework works to enable synchronization of your Systems of Record (SoR). In addition, we’ve also shown a Nautobot SSoT App for Arista CloudVision which extends the SSoT base framework. So now you ask, how do I synchronize my data to and from Nautobot using the Single Source of Truth framework? In this first part of a two-part series, I’ll be explaining the basics of creating your own SSoT app, and then next month, I’ll be following up with more advanced options available when building an SSoT app.

Please note: it is expected that you’ve read the Nautobot Plugin: Single Source of Truth (SSoT) post and understand the framework terminology, such as Data Source and Data Target.

The first thing to do when creating an SSoT app is to define the data shared between your SoR that you want to synchronize. For example, you might want to pull your Rooms from Device42 into Nautobot. You would then create a class that inherits from the DiffSyncModel class as shown below:

from diffsync import DiffSyncModel
from typing import List, Optional


class Room(DiffSyncModel):
    """Room model."""

    _modelname = "room"
    _identifiers = ("name", "building")
    _shortname = ("name",)
    _attributes = ("notes",)
    _children = {"rack": "racks"}
    name: str
    building: str
    notes: Optional[str]
    racks: List["Rack"] = list()

You’ll notice that there are both public and private class attributes defined. Each private attribute is used to help define the model itself within the DiffSync framework. There are two attributes required on every DiffSyncModel, the _modelname and _identifiers attributes. The _modelname attribute defines the type of the model and is used to identify the shared models between your SoR. The _identifier attribute specifies the public attributes used to generate a name for the objects created when loading data from your adapters. It’s essential to confirm that the identifiers used for the object make it globally unique to ensure an accurate sync.

The remaining attributes are optional but can be quite useful in the process. The _shortname attribute identifies an object apart from other objects of the same type allowing for use of a shorter name. The _attributes attribute specifies all attributes that are of interest for synchronization. You’ll notice that each of these public attributes is defined using pydantic typing syntax. These are essential for ensuring data integrity while performing the synchronization. Please note that you must use the Optional type for any attribute that you wish to allow to be None. The last private attribute is _children, which defines other models related to the model you’re creating. In this example, Rooms are children of the Building as you have many Rooms inside a Building. This allows you to define a hierarchy of models for importing. Please note that this is meant for a direct parent-to-child relationship and not multi-branching inheritance. The _children attribute is defined using the pattern of {<model_name>: <field_name>}.

The next step is to define the CRUD (Create, Update, Delete) methods for each model. These methods will handle taking the data, once loaded from your Data Source, and making the relevant changes to the object in your Data Target. Although you may add the CRUD methods for your object to the DiffSyncModel class that you created in the first step, best practice is to create new classes that inherit from that DiffSyncModel class, as shown below:

from django.utils.text import slugify
from nautobot.dcim.models import RackGroup as NautobotRackGroup
from nautobot.dcim.models import Site as NautobotSite


class NautobotRoom(Room):
    
    """Nautobot Room CRUD methods."""

    @classmethod
    def create(cls, diffsync, ids, attrs):
        """Create RackGroup object in Nautobot."""
        new_rg = NautobotRackGroup(
            name=ids["name"],
            slug=slugify(ids["name"]),
            site=NautobotSite.objects.get(name=ids["building"]),
            description=attrs["notes"] if attrs.get("notes") else "",
        )
        new_rg.validated_save()
        return super().create(ids=ids, diffsync=diffsync, attrs=attrs)

    def update(self, attrs):
        """Update RackGroup object in Nautobot."""
        _rg = NautobotRackGroup.objects.get(name=self.name, site__name=self.building)
        if attrs.get("notes"):
            _rg.description = attrs["notes"]
        _rg.validated_save()
        return super().update(attrs)

    def delete(self):
        """Delete RackGroup object from Nautobot."""
        self.diffsync.job.log_warning(f"RackGroup {self.name} will be deleted.")
        super().delete()
        rackgroup = NautobotRackGroup.objects.get(**self.get_identifiers())
        rackgroup.delete()
        return self

Each of the create(), update(), and delete() methods for an object are called once a diff is completed and the synchronization process is started. Which method is called depends upon the required changes to the object in your Data Target. When the create() method is called, the object’s identifier and other attributes are passed to it as the ids and attrs variables respectively. The diffsync variable is for handling interactions with the DiffSync Job, such as sending log messages. For the logging of the Job results to be accurate, it is essential that the object is returned to the create method with the variables passed. However, unlike the create() method, the update() method receives only the attributes that have been changed for an object. This means that it is required for the implementer to validate if attributes have been passed or not before making appropriate changes. The delete() method will receive only the class object itself.

When utilizing inheritance between models, ensure the related models have the update_forward_refs() method called. This is essential to establish the relationships between objects.

Once the models and their CRUD methods have been defined, the next step is to write the adapters that load the models you specified in the previous steps. It is in this sense that the adapter class adapts the data from your Data Source. The adapters are required to reference each model that you wish to have considered at the top of the DiffSync object along with a top_level list of your models in the order that you wish to have them processed, as you can see below:

from diffsync import DiffSync
from .models import Building, Room


class Device42Adapter(DiffSync):
    """DiffSync adapter for Device42 server."""

    building = Building
    room = Room

    top_level = ["building"]

from diffsync import DiffSync
from .models import Building, Room


class NautobotAdapter(DiffSync):
    """Nautobot adapter for DiffSync."""

    building = Building
    room = Room

    top_level = ["building"]

As you can see above, you will always have two Systems of Record in a diff so you will need an adapter for both. It is best practice to have them matching at the top to ensure that items are processed identically. As you can see in the examples above, only the Building model is in the top_level list as the Room model is a child and will be processed after the Building is. It is up to the implementer to determine how they wish to load the models they create in the adapters. While loading your models from the methods in your adapters, it is essential that you pass valid DiffSyncModel objects that adhere to what you specified in your models when passed to the add() function. Failing to do so will cause validation errors.

It’s advised to use a load() method to call your other model-specific methods to keep things concise.

from diffsync.exceptions import ObjectAlreadyExists

class Device42Adapter(DiffSync):
...

    def load_rooms(self):
        """Load Device42 rooms."""
        for record in self._device42.api_call(path="api/1.0/rooms")["rooms"]:
            if record.get("building"):
                room = self.room(
                    name=record["name"],
                    building=record["building"],
                    notes=record["notes"] if record.get("notes") else "",
                )
                try:
                    self.add(room)
                    _site = self.get(self.building, record.get("building"))
                    _site.add_child(child=room)
                except ObjectAlreadyExists as err:
                    self.job.log_warning(f"{record['name']} is already loaded. {err}")
            else:
                self.job.log_warning(f"{record['name']} missing Building, skipping.")
                continue

The example above shows how data is pulled from the Device42 API and creates the Room objects that were detailed in the first step. Once the object has been created, it is then added into the DiffSync set with the add() method. As a Room is a child object of a Building, there is an additional step of finding the parent Building object with the get() method, and then using the add_child() method to add the relationship between the objects. If there is an existing object with the same identifiers, the ObjectAlreadyExists exception will be thrown, so it’s advised to wrap the add() method in a try/except block.

With your adapters for each SoR created, the final step is to write your Nautobot Job. This will handle the loading of your models from the adapters, the diff of the objects once loaded, and the synchronization of data by calling the CRUD methods as appropriate. The Job class must be derived from either the DataSource or DataTarget class and is required to include a sync_data method to handle the synchronization process. Optionally, you can also add a config_information or data_mappings method to enrich the data presented to the end user in Nautobot.

from django.templatetags.static import static
from nautobot.extras.jobs import Job
from nautobot_ssot.jobs.base import DataSource

from diffsync.exceptions import ObjectNotCreated
from .device42 import Device42Adapter
from .nautobot import NautobotAdapter


class Device42DataSource(DataSource, Job):
    """Device42 SSoT Data Source."""

    class Meta:
        """Meta data for Device42."""

        name = "Device42"
        data_source = "Device42"
        data_source_icon = static("./d42_logo.png")
        description = "Sync information from Device42 to Nautobot"

    def sync_data(self):
        """Device42 Sync."""
        d42_adapter = Device42Adapter(job=self, sync=self.sync)
        d42_adapter.load()
        nb_adapter = NautobotAdapter(job=self, sync=self.sync)
        nb_adapter.load()
        diff = nb_adapter.diff_from(d42_adapter)
        self.sync.diff = diff.dict()
        self.sync.save()
        self.log_info(message=diff.summary())
        if not self.kwargs["dry_run"]:
            try:
                nb_adapter.sync_from(d42_adapter)
            except ObjectNotCreated as err:
                self.log_debug(f"Unable to create object. {err}")
            self.log_success(message="Sync complete.")


jobs = [Device42DataSource]

As is shown, the Job should create an instance of each of your adapters and call their load() methods to create the DiffSyncModel objects. Once that’s done, a diff and sync can be completed utilizing either the diff_from/diff_to and sync_from/sync_to methods on the adapter objects. Which you use depends upon which way you wish to have the synchronization performed. In the example above, once the models have been loaded, a diff from Device42 to Nautobot is done to report the required objects to be created, updated, or deleted. Finally, if the Job is not a dry-run, synchronization will be executed. Again, this is done with a sync_from from the Device42 adapter object to the Nautobot adapter object. Depending upon the results of an object being created, an ObjectNotCreated exception may be thrown, so it’s advised to use a try/except block when calling the sync methods to ensure it’s caught and handled appropriately. Once all of the objects have been processed, a final success log is sent and the GUI should be updated to reflect the changes.

In summary, the creation of an SSoT app requires the following steps to be completed:

1. Create one or more DiffSyncModel classes to define the data you wish to synchronize.
2. For each DiffSyncModel, define the CRUD methods to handle the requisite changes in your Data Target.
3. Write a DiffSync adapter class for each System of Record to load data from each into your DiffSyncModel classes.
4. Finally, write your Nautobot Job to perform the synchronization of data between your Data Source and Data Target.