Code reference

This page aims to provide an API overview of the different functions and classes you may need when implementing an SSoT job.

nautobot_ssot.contrib.CustomFieldAnnotation dataclass

Map a model field to an arbitrary custom field name.

For usage with typing.Annotated.

This exists to map model fields to their corresponding custom fields. This serves to explicitly differentiate normal fields from custom fields.

Note that for backwards compatibility purposes it is also possible to use CustomFieldAnnotation.name instead of CustomFieldAnnotation.key.

Example

Given a boolean custom field with label "Is Global" and key "is_global" on the Provider model:

class ProviderModel(NautobotModel):
    _model: Provider
    _identifiers = ("name",)
    _attributes = ("is_global",)

    name: str
    is_global: Annotated[bool, CustomFieldAnnotation(key="is_global")

This then maps the model field 'is_global' to the custom field with the key 'is_global'.

Source code in nautobot_ssot/contrib/types.py

@dataclass
class CustomFieldAnnotation:
    """Map a model field to an arbitrary custom field name.

    For usage with `typing.Annotated`.

    This exists to map model fields to their corresponding custom fields. This serves to explicitly differentiate
    normal fields from custom fields.

    Note that for backwards compatibility purposes it is also possible to use `CustomFieldAnnotation.name` instead of
    `CustomFieldAnnotation.key`.

    Example:
        Given a boolean custom field with label "Is Global" and key "is_global" on the Provider model:

        ```python
        class ProviderModel(NautobotModel):
            _model: Provider
            _identifiers = ("name",)
            _attributes = ("is_global",)

            name: str
            is_global: Annotated[bool, CustomFieldAnnotation(key="is_global")
        ```

        This then maps the model field 'is_global' to the custom field with the key 'is_global'.
    """

    # TODO: Delete on 3.0, keep around for backwards compatibility for now
    name: Optional[str] = None

    key: Optional[str] = None

    def __post_init__(self):
        """Compatibility layer with using 'name' instead of 'key'.

        If `self.key` isn't set, fall back to the old behaviour.
        """
        if not self.key:
            if self.name:
                self.key = self.name
            else:
                raise ValueError("The 'key' field on CustomFieldAnnotation needs to be set.")

__post_init__()

Compatibility layer with using 'name' instead of 'key'.

If self.key isn't set, fall back to the old behaviour.

Source code in nautobot_ssot/contrib/types.py

def __post_init__(self):
    """Compatibility layer with using 'name' instead of 'key'.

    If `self.key` isn't set, fall back to the old behaviour.
    """
    if not self.key:
        if self.name:
            self.key = self.name
        else:
            raise ValueError("The 'key' field on CustomFieldAnnotation needs to be set.")

nautobot_ssot.contrib.NautobotModel

Bases: DiffSyncModel, BaseNautobotModel

Base model for any diffsync models interfacing with Nautobot through the ORM.

This provides the create, update and delete operations in a generic fashion, meaning you don't have to implement them yourself.

In order to accomplish this, the _model field has to be set on subclasses to map them to the corresponding ORM model class.

Source code in nautobot_ssot/contrib/model.py

class NautobotModel(DiffSyncModel, BaseNautobotModel):
    """
    Base model for any diffsync models interfacing with Nautobot through the ORM.

    This provides the `create`, `update` and `delete` operations in a generic fashion, meaning you don't have to
    implement them yourself.

    In order to accomplish this, the `_model` field has to be set on subclasses to map them to the corresponding ORM
    model class.
    """

    @classmethod
    def get_synced_attributes(cls) -> List[str]:
        """Return a list of parameters synced as part of the SSoT Process."""
        return list(cls._identifiers) + list(cls._attributes)

    @classmethod
    def _get_queryset(cls) -> QuerySet:
        """Get the queryset used to load the models data from Nautobot."""
        available_fields = {field.name for field in cls._model._meta.get_fields()}
        parameter_names = [
            parameter for parameter in cls.get_synced_attributes() if parameter.split("__")[0] in available_fields
        ]
        # Here we identify any foreign keys (i.e. fields with '__' in them) so that we can load them directly in the
        # first query if this function hasn't been overridden.
        prefetch_related_parameters = [
            "__".join(parameter.split("__")[:-1]) for parameter in parameter_names if "__" in parameter
        ]
        qs = cls.get_queryset()
        return qs.prefetch_related(*prefetch_related_parameters)

    @classmethod
    def get_queryset(cls) -> QuerySet:
        """Get the queryset used to load the models data from Nautobot."""
        return cls._model.objects.all()

    @classmethod
    def _check_field(cls, name):
        """Check whether the given field name is defined on the diffsync (pydantic) model."""
        if name not in cls.model_fields:  # pylint: disable=unsupported-membership-test
            raise ObjectCrudException(f"Field {name} is not defined on the model.")

    def get_from_db(self):
        """Get the ORM object for this diffsync object from the database using the primary key."""
        try:
            return self.adapter.get_from_orm_cache({"pk": self.pk}, self._model)
        except self._model.DoesNotExist as error:
            raise ObjectCrudException(f"No such {self._model._meta.verbose_name} instance with PK {self.pk}") from error

    def update(self, attrs):
        """Update the ORM object corresponding to this diffsync object."""
        try:
            obj = self.get_from_db()
            self._update_obj_with_parameters(obj, attrs, self.adapter)
            if getattr(self.adapter, "metadata_type", None):
                self._update_obj_metadata(obj, self.adapter)
        except ObjectCrudException as error:
            raise ObjectNotUpdated(error) from error
        return super().update(attrs)

    def delete(self):
        """Delete the ORM object corresponding to this diffsync object."""
        try:
            obj = self.get_from_db()
        except ObjectCrudException as error:
            raise ObjectNotDeleted(error) from error
        try:
            obj.delete()
        except ProtectedError as error:
            raise ObjectNotDeleted(f"Couldn't delete {obj} as it is referenced by another object") from error
        return super().delete()

    @classmethod
    def create(cls, adapter, ids, attrs):
        """Create the ORM object corresponding to this diffsync object."""
        # Only diffsync cares about the distinction between ids and attrs, we do not.
        # Therefore, we merge the two into parameters.
        parameters = ids.copy()
        parameters.update(attrs)

        # This is in fact callable, because it is a model
        obj = cls._model()  # pylint: disable=not-callable

        try:
            cls._update_obj_with_parameters(obj, parameters, adapter)
        except ObjectCrudException as error:
            raise ObjectNotCreated(error) from error

        if getattr(adapter, "metadata_type", None):
            try:
                cls._update_obj_metadata(obj, adapter)
            except ObjectCrudException as error:
                raise ObjectNotCreated(error) from error

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

    @classmethod
    def _handle_single_field(cls, field, obj, value, relationship_fields, adapter):  # pylint: disable=too-many-arguments,too-many-locals, too-many-branches
        """Set a single field on a Django object to a given value, or, for relationship fields, prepare setting.

        :param field: The name of the field to set.
        :param obj: The Django ORM object to set the field on.
        :param value: The value to set the field to.
        :param relationship_fields: Helper dictionary containing information on relationship fields.
            This is mutated over the course of this function.
        :param adapter: The related diffsync adapter used for looking up things in the cache.
        """
        # Use type hints at runtime to determine which fields are custom fields
        type_hints = get_type_hints(cls, include_extras=True)

        cls._check_field(field)

        # Handle custom fields. See CustomFieldAnnotation docstring for more details.
        custom_relationship_annotation = None
        metadata_for_this_field = getattr(type_hints[field], "__metadata__", [])
        for metadata in metadata_for_this_field:
            if isinstance(metadata, CustomFieldAnnotation):
                obj.cf[metadata.key] = value
                return
            if isinstance(metadata, CustomRelationshipAnnotation):
                custom_relationship_annotation = metadata
                break

        # Prepare handling of foreign keys and custom relationship foreign keys.
        # Example: If field is `tenant__group__name`, then
        # `foreign_keys["tenant"]["group__name"] = value` or
        # `custom_relationship_foreign_keys["tenant"]["group__name"] = value`
        # Also, the model class will be added to the dictionary for normal foreign keys, so we can later use it
        # for querying:
        # `foreign_keys["tenant"]["_model_class"] = nautobot.tenancy.models.Tenant
        # For custom relationship foreign keys, we add the annotation instead:
        # `custom_relationship_foreign_keys["tenant"]["_annotation"] = CustomRelationshipAnnotation(...)
        if "__" in field:
            related_model, lookup = field.split("__", maxsplit=1)
            # Custom relationship foreign keys
            if custom_relationship_annotation:
                relationship_fields["custom_relationship_foreign_keys"][related_model][lookup] = value
                relationship_fields["custom_relationship_foreign_keys"][related_model]["_annotation"] = (
                    custom_relationship_annotation
                )
            # Normal foreign keys
            else:
                django_field = cls._model._meta.get_field(related_model)
                relationship_fields["foreign_keys"][related_model][lookup] = value
                # Add a special key to the dictionary to point to the related model's class
                relationship_fields["foreign_keys"][related_model]["_model_class"] = django_field.related_model
            return

        # Prepare handling of custom relationship many-to-many fields.
        if custom_relationship_annotation:
            relationship = adapter.get_from_orm_cache({"label": custom_relationship_annotation.name}, Relationship)
            if custom_relationship_annotation.side == RelationshipSideEnum.DESTINATION:
                related_object_content_type = relationship.source_type
            else:
                related_object_content_type = relationship.destination_type
            related_model_class = related_object_content_type.model_class()
            if (
                relationship.type == RelationshipTypeChoices.TYPE_ONE_TO_MANY
                and custom_relationship_annotation.side == RelationshipSideEnum.DESTINATION
            ):
                relationship_fields["custom_relationship_foreign_keys"][field] = {
                    **value,
                    "_annotation": custom_relationship_annotation,
                }
            else:
                relationship_fields["custom_relationship_many_to_many_fields"][field] = {
                    "annotation": custom_relationship_annotation,
                    "objects": [adapter.get_from_orm_cache(parameters, related_model_class) for parameters in value],
                }

            return

        django_field = cls._model._meta.get_field(field)

        # Prepare handling of many-to-many fields. If we are dealing with a many-to-many field,
        # we get all the related objects here to later set them once the object has been saved.
        if django_field.many_to_many or django_field.one_to_many:
            try:
                relationship_fields["many_to_many_fields"][field] = [
                    adapter.get_from_orm_cache(parameters, django_field.related_model) for parameters in value
                ]
            except django_field.related_model.DoesNotExist as error:
                raise ObjectCrudException(
                    f"Unable to populate many to many relationship '{django_field.name}' with parameters {value}, at least one related object not found."
                ) from error
            except MultipleObjectsReturned as error:
                raise ObjectCrudException(
                    f"Unable to populate many to many relationship '{django_field.name}' with parameters {value}, at least one related object found twice."
                ) from error
            return

        # As the default case, just set the attribute directly
        setattr(obj, field, value)

    @classmethod
    def _update_obj_with_parameters(cls, obj, parameters, adapter):
        """Update a given Nautobot ORM object with the given parameters."""
        relationship_fields = {
            # Example: {"group": {"name": "Group Name", "_model_class": TenantGroup}}
            "foreign_keys": defaultdict(dict),
            # Example: {"tags": [Tag-1, Tag-2]}
            "many_to_many_fields": defaultdict(list),
            # Example: TODO
            "custom_relationship_foreign_keys": defaultdict(dict),
            # Example: TODO
            "custom_relationship_many_to_many_fields": defaultdict(dict),
        }
        for field, value in parameters.items():
            cls._handle_single_field(field, obj, value, relationship_fields, adapter)

        # Set foreign keys
        cls._lookup_and_set_foreign_keys(relationship_fields["foreign_keys"], obj, adapter)

        # Save the object to the database
        try:
            obj.validated_save()
        except (ValidationError, ValueError) as error:
            raise ObjectCrudException(
                f"Validated save failed for Django object:\n{error}\nParameters: {parameters}"
            ) from error

        # Handle relationship association creation. This needs to be after object creation, because relationship
        # association objects rely on both sides already existing.
        cls._lookup_and_set_custom_relationship_foreign_keys(
            relationship_fields["custom_relationship_foreign_keys"], obj, adapter
        )
        cls._set_custom_relationship_to_many_fields(
            relationship_fields["custom_relationship_many_to_many_fields"], obj, adapter
        )

        # Set many-to-many fields after saving.
        cls._set_many_to_many_fields(relationship_fields["many_to_many_fields"], obj)

    @classmethod
    def _set_custom_relationship_to_many_fields(cls, custom_relationship_many_to_many_fields, obj, adapter):
        for _, dictionary in custom_relationship_many_to_many_fields.items():
            annotation = dictionary.pop("annotation")
            objects = dictionary.pop("objects")
            # TODO: Deduplicate this code
            relationship = adapter.get_from_orm_cache({"label": annotation.name}, Relationship)
            parameters = {
                "relationship": relationship,
                "source_type": relationship.source_type,
                "destination_type": relationship.destination_type,
            }
            associations = []
            if annotation.side == RelationshipSideEnum.SOURCE:
                parameters["source_id"] = obj.id
                for object_to_relate in objects:
                    association_parameters = parameters.copy()
                    association_parameters["destination_id"] = object_to_relate.id
                    try:
                        association = adapter.get_from_orm_cache(association_parameters, RelationshipAssociation)
                    except RelationshipAssociation.DoesNotExist:
                        association = RelationshipAssociation(**parameters, destination_id=object_to_relate.id)
                        association.validated_save()
                    associations.append(association)
            else:
                parameters["destination_id"] = obj.id
                for object_to_relate in objects:
                    association_parameters = parameters.copy()
                    association_parameters["source_id"] = object_to_relate.id
                    try:
                        association = adapter.get_from_orm_cache(association_parameters, RelationshipAssociation)
                    except RelationshipAssociation.DoesNotExist:
                        association = RelationshipAssociation(**parameters, source_id=object_to_relate.id)
                        association.validated_save()
                    associations.append(association)
            # Now we need to clean up any associations that we're not `get_or_create`'d in order to achieve
            # declarativeness.
            # TODO: This may benefit from an ORM cache with `filter` capabilities, but I guess the gain in most cases
            # would be fairly minor.
            for existing_association in RelationshipAssociation.objects.filter(**parameters):
                if existing_association not in associations:
                    existing_association.delete()

    @classmethod
    def _set_many_to_many_fields(cls, many_to_many_fields, obj):
        """
        Given a dictionary, set many-to-many relationships on an object.

        This will always use `set`, thereby replacing any elements that are already part of the relationship.

        Example dictionary:
        {
            "relationship_field_name": [<Object 1>, <Object 2>],
            ...
        }
        """
        for field_name, related_objects in many_to_many_fields.items():
            many_to_many_field = getattr(obj, field_name)
            many_to_many_field.set(related_objects)

    @classmethod
    def _lookup_and_set_custom_relationship_foreign_keys(cls, custom_relationship_foreign_keys, obj, adapter):
        for _, related_model_dict in custom_relationship_foreign_keys.items():
            annotation = related_model_dict.pop("_annotation")
            # TODO: Deduplicate this code
            try:
                relationship = adapter.get_from_orm_cache({"label": annotation.name}, Relationship)
            except Relationship.DoesNotExist as error:
                raise ObjectCrudException(f"No such relationship with label '{annotation.name}'") from error
            parameters = {
                "relationship": relationship,
                "source_type": relationship.source_type,
                "destination_type": relationship.destination_type,
            }
            if annotation.side == RelationshipSideEnum.SOURCE:
                parameters["source_id"] = obj.id
                related_model_class = relationship.destination_type.model_class()
                try:
                    destination_object = adapter.get_from_orm_cache(related_model_dict, related_model_class)
                except related_model_class.DoesNotExist as error:
                    raise ObjectCrudException(
                        f"Couldn't resolve custom relationship {relationship.name}, no such {related_model_class._meta.verbose_name} object with parameters {related_model_dict}."
                    ) from error
                except related_model_class.MultipleObjectsReturned as error:
                    raise ObjectCrudException(
                        f"Couldn't resolve custom relationship {relationship.name}, multiple {related_model_class._meta.verbose_name} objects with parameters {related_model_dict}."
                    ) from error
                RelationshipAssociation.objects.update_or_create(
                    **parameters,
                    defaults={"destination_id": destination_object.id},
                )
            else:
                parameters["destination_id"] = obj.id
                source_object = adapter.get_from_orm_cache(related_model_dict, relationship.source_type.model_class())
                RelationshipAssociation.objects.update_or_create(**parameters, defaults={"source_id": source_object.id})

    @classmethod
    def _lookup_and_set_foreign_keys(cls, foreign_keys, obj, adapter):
        """
        Given a list of foreign keys as dictionaries, look up and set foreign keys on an object.

        Dictionary should be in the form of:
        [
          {"field_1": "value_1", "field_2": "value_2"},
          ...
        ]
        where each item in the list corresponds to the parameters needed to uniquely identify a foreign key object.
        """
        for field_name, related_model_dict in foreign_keys.items():
            related_model = related_model_dict.pop("_model_class")
            # Generic foreign keys will not have this dictionary field. As such, we need to retrieve the appropriate
            # model class through other means.
            if not related_model:
                try:
                    app_label = related_model_dict.pop("app_label")
                    model = related_model_dict.pop("model")
                except KeyError as error:
                    raise ValueError(
                        f"Missing annotation for '{field_name}__app_label' or '{field_name}__model - this is required"
                        f"for generic foreign keys."
                    ) from error
                try:
                    related_model_content_type = adapter.get_from_orm_cache(
                        {"app_label": app_label, "model": model}, ContentType
                    )
                    related_model = related_model_content_type.model_class()
                except ContentType.DoesNotExist as error:
                    raise ObjectCrudException(f"Unknown content type '{app_label}.{model}'.") from error
            # Set the foreign key to 'None' when none of the fields are set to anything
            if not any(related_model_dict.values()):
                setattr(obj, field_name, None)
                continue
            try:
                related_object = adapter.get_from_orm_cache(related_model_dict, related_model)
            except related_model.DoesNotExist as error:
                raise ObjectCrudException(
                    f"Couldn't find '{related_model._meta.verbose_name}' instance behind '{field_name}' with: {related_model_dict}."
                ) from error
            except MultipleObjectsReturned as error:
                raise ObjectCrudException(
                    f"Found multiple instances for {field_name} with: {related_model_dict}"
                ) from error
            setattr(obj, field_name, related_object)

    @classmethod
    def _update_obj_metadata(cls, obj, adapter):
        """Update a given Nautobot ORM object with the required object metadata."""
        # Get the scope_fields from the DiffSync Model
        obj_metadata_scope_fields = adapter.metadata_scope_fields[cls]
        obj_metadata = obj.associated_object_metadata.filter(
            metadata_type=adapter.metadata_type
        ).first() or ObjectMetadata(metadata_type=adapter.metadata_type, assigned_object=obj)

        obj_metadata.scoped_fields = obj_metadata_scope_fields
        obj_metadata.value = datetime.now()
        obj_metadata.validated_save()

create(adapter, ids, attrs) classmethod

Create the ORM object corresponding to this diffsync object.

Source code in nautobot_ssot/contrib/model.py

@classmethod
def create(cls, adapter, ids, attrs):
    """Create the ORM object corresponding to this diffsync object."""
    # Only diffsync cares about the distinction between ids and attrs, we do not.
    # Therefore, we merge the two into parameters.
    parameters = ids.copy()
    parameters.update(attrs)

    # This is in fact callable, because it is a model
    obj = cls._model()  # pylint: disable=not-callable

    try:
        cls._update_obj_with_parameters(obj, parameters, adapter)
    except ObjectCrudException as error:
        raise ObjectNotCreated(error) from error

    if getattr(adapter, "metadata_type", None):
        try:
            cls._update_obj_metadata(obj, adapter)
        except ObjectCrudException as error:
            raise ObjectNotCreated(error) from error

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

delete()

Delete the ORM object corresponding to this diffsync object.

Source code in nautobot_ssot/contrib/model.py

def delete(self):
    """Delete the ORM object corresponding to this diffsync object."""
    try:
        obj = self.get_from_db()
    except ObjectCrudException as error:
        raise ObjectNotDeleted(error) from error
    try:
        obj.delete()
    except ProtectedError as error:
        raise ObjectNotDeleted(f"Couldn't delete {obj} as it is referenced by another object") from error
    return super().delete()

get_from_db()

Get the ORM object for this diffsync object from the database using the primary key.

Source code in nautobot_ssot/contrib/model.py

def get_from_db(self):
    """Get the ORM object for this diffsync object from the database using the primary key."""
    try:
        return self.adapter.get_from_orm_cache({"pk": self.pk}, self._model)
    except self._model.DoesNotExist as error:
        raise ObjectCrudException(f"No such {self._model._meta.verbose_name} instance with PK {self.pk}") from error

get_queryset() classmethod

Get the queryset used to load the models data from Nautobot.

Source code in nautobot_ssot/contrib/model.py

@classmethod
def get_queryset(cls) -> QuerySet:
    """Get the queryset used to load the models data from Nautobot."""
    return cls._model.objects.all()

get_synced_attributes() classmethod

Return a list of parameters synced as part of the SSoT Process.

Source code in nautobot_ssot/contrib/model.py

@classmethod
def get_synced_attributes(cls) -> List[str]:
    """Return a list of parameters synced as part of the SSoT Process."""
    return list(cls._identifiers) + list(cls._attributes)

update(attrs)

Update the ORM object corresponding to this diffsync object.

Source code in nautobot_ssot/contrib/model.py

def update(self, attrs):
    """Update the ORM object corresponding to this diffsync object."""
    try:
        obj = self.get_from_db()
        self._update_obj_with_parameters(obj, attrs, self.adapter)
        if getattr(self.adapter, "metadata_type", None):
            self._update_obj_metadata(obj, self.adapter)
    except ObjectCrudException as error:
        raise ObjectNotUpdated(error) from error
    return super().update(attrs)

nautobot_ssot.contrib.NautobotAdapter

Bases: Adapter, BaseNautobotAdapter

Adapter for loading data from Nautobot through the ORM.

This adapter is able to infer how to load data from Nautobot based on how the models attached to it are defined.

Source code in nautobot_ssot/contrib/adapter.py

class NautobotAdapter(Adapter, BaseNautobotAdapter):
    """
    Adapter for loading data from Nautobot through the ORM.

    This adapter is able to infer how to load data from Nautobot based on how the models attached to it are defined.
    """

    def __init__(self, *args, job, sync=None, **kwargs):
        """Instantiate this class, but do not load data immediately from the local system."""
        super().__init__(*args, **kwargs)
        self.job = job
        self.sync = sync
        self.cache: ORMCache = kwargs.pop("cache", ORMCache())
        self.metadata_type = None
        self.metadata_scope_fields = {}

    def get_from_orm_cache(self, parameters: Dict, model_class: Type[Model]):
        """Retrieve an object from the ORM or the cache."""
        return self.cache.get_from_orm(model_class, parameters)

    @staticmethod
    def _get_parameter_names(diffsync_model):
        """Ignore the differences between identifiers and attributes, because at this point they don't matter to us."""
        return list(diffsync_model._identifiers) + list(diffsync_model._attributes)  # pylint: disable=protected-access

    def invalidate_cache(self, zero_out_hits=True):
        """Deprecated, kept for backwards compatibility."""
        self.job.logger.warning(
            "Adapter class method `self.invalidate_cache()` is deprecated and will be removed in a future version. "
            "Use `self.cache.invalidate_cache()` instead."
        )
        self.cache.invalidate_cache(zero_out_hits=zero_out_hits)

    def _load_objects(self, diffsync_model: BaseNautobotModel):
        """Given a diffsync model class, load a list of models from the database and return them."""
        parameter_names = diffsync_model.get_synced_attributes()
        for database_object in diffsync_model._get_queryset():
            self._load_single_object(database_object, diffsync_model, parameter_names)

    def _handle_single_parameter(self, parameters, parameter_name, database_object, diffsync_model):
        type_hints = get_type_hints(diffsync_model, include_extras=True)
        # Handle custom fields and custom relationships. See CustomFieldAnnotation and CustomRelationshipAnnotation
        # docstrings for more details.
        is_custom_field = False
        custom_relationship_annotation = None
        metadata_for_this_field = getattr(type_hints[parameter_name], "__metadata__", [])
        for metadata in metadata_for_this_field:
            if isinstance(metadata, CustomFieldAnnotation):
                field_key = metadata.key or metadata.name
                if field_key in database_object.cf:
                    parameters[parameter_name] = database_object.cf[field_key]
                is_custom_field = True
                break
            if isinstance(metadata, CustomRelationshipAnnotation):
                custom_relationship_annotation = metadata
                break
        if is_custom_field:
            return

        # Handling of foreign keys where the local side is the many and the remote side the one.
        # Note: This includes the side of a generic foreign key that has the foreign key, i.e.
        # the 'many' side.
        if "__" in parameter_name:
            if custom_relationship_annotation:
                parameters[parameter_name] = self._handle_custom_relationship_foreign_key(
                    database_object, parameter_name, custom_relationship_annotation
                )
            else:
                parameters[parameter_name] = orm_attribute_lookup(database_object, parameter_name)
            return

        # Handling of one- and many-to custom relationship fields:
        if custom_relationship_annotation:
            parameters[parameter_name] = self._handle_custom_relationship_to_many_relationship(
                database_object, diffsync_model, parameter_name, custom_relationship_annotation
            )
            return

        database_field = diffsync_model._model._meta.get_field(parameter_name)

        # Handling of one- and many-to-many non-custom relationship fields.
        # Note: This includes the side of a generic foreign key that constitutes the foreign key,
        # i.e. the 'one' side.
        if database_field.many_to_many or database_field.one_to_many:
            parameters[parameter_name] = self._handle_to_many_relationship(
                database_object, diffsync_model, parameter_name
            )
            return

        # Handling of normal fields - as this is the default case, set the attribute directly.
        if hasattr(self, f"load_param_{parameter_name}"):
            parameters[parameter_name] = getattr(self, f"load_param_{parameter_name}")(parameter_name, database_object)
        else:
            parameters[parameter_name] = getattr(database_object, parameter_name)

    def _load_single_object(self, database_object, diffsync_model, parameter_names):
        """Load a single diffsync object from a single database object."""
        parameters = {}
        for parameter_name in parameter_names:
            self._handle_single_parameter(parameters, parameter_name, database_object, diffsync_model)
        parameters["pk"] = database_object.pk
        try:
            diffsync_model = diffsync_model(**parameters)
        except pydantic.ValidationError as error:
            raise ValueError(f"Parameters: {parameters}") from error
        self.add(diffsync_model)

        self._handle_children(database_object, diffsync_model)
        return diffsync_model

    def _handle_children(self, database_object, diffsync_model: BaseNautobotModel):
        """Recurse through all the children for this model."""
        for children_parameter, children_field in diffsync_model._children.items():
            children = getattr(database_object, children_field).all()
            diffsync_model_child: BaseNautobotModel = self._get_diffsync_class(model_name=children_parameter)
            for child in children:
                parameter_names = diffsync_model_child.get_synced_attributes()
                child_diffsync_object = self._load_single_object(child, diffsync_model_child, parameter_names)
                diffsync_model.add_child(child_diffsync_object)

    def load(self):
        """Generic implementation of the load function."""
        if not hasattr(self, "top_level") or not self.top_level:
            raise ValueError("'top_level' needs to be set on the class.")

        for model_name in self.top_level:
            diffsync_model = self._get_diffsync_class(model_name)

            # This function directly mutates the diffsync store, i.e. it will create and load the objects
            # for this specific model class as well as its children without returning anything.
            self._load_objects(diffsync_model)

    def _get_diffsync_class(self, model_name):
        """Given a model name, return the diffsync class."""
        try:
            diffsync_model = getattr(self, model_name)
        except AttributeError as error:
            raise AttributeError(
                f"Please define {model_name} to be the diffsync model on this adapter as a class level attribute."
            ) from error
        return diffsync_model

    def _handle_custom_relationship_to_many_relationship(
        self,
        database_object: Model,
        diffsync_model: DiffSyncModel,
        parameter_name: str,
        annotation: CustomRelationshipAnnotation,
    ):
        # Introspect type annotations to deduce which fields are of interest
        # for this many-to-many relationship.
        inner_type = get_inner_type(diffsync_model, parameter_name)
        # TODO: Allow for filtering, i.e. not taking into account all the objects behind the relationship.
        relationship: Relationship = self.cache.get_from_orm(Relationship, {"label": annotation.name})
        relationship_associations, _ = get_custom_relationship_associations(
            relationship=relationship,
            db_obj=database_object,
            relationship_side=annotation.side,
        )

        related_objects_list = []
        for association in relationship_associations:
            related_object = getattr(
                association, "source" if annotation.side == RelationshipSideEnum.DESTINATION else "destination"
            )
            dictionary_representation = load_typed_dict(inner_type, related_object)
            # Only use those where there is a single field defined, all 'None's will not help us.
            if any(dictionary_representation.values()):
                related_objects_list.append(dictionary_representation)

        # For one-to-many, we need to return an object, not a list of objects
        if (
            relationship.type == RelationshipTypeChoices.TYPE_ONE_TO_MANY
            and annotation.side == RelationshipSideEnum.DESTINATION
        ):
            if not related_objects_list:
                return None

            if len(related_objects_list) == 1:
                return related_objects_list[0]

            raise ObjectCrudException(
                f"More than one related objects for a {RelationshipTypeChoices.TYPE_ONE_TO_MANY} relationship: {related_objects_list}"
            )

        return related_objects_list

    def _handle_to_many_relationship(self, database_object, diffsync_model, parameter_name):
        """Handle a single one- or many-to-many relationship field.

        one- or many-to-many relationships are type annotated as a list of typed dictionaries. The typed
        dictionary type expresses, which attributes we are interested in for diffsync.

        :param database_object: The Django ORM database object
        :param diffsync_model: The diffsync model class (not specific object) for this ORM object
        :param parameter_name: The field name of the specific relationship to handle
        :return: A list of dictionaries which represent the related objects.

        :example:

        Example parameters:
        - a `nautobot.dcim.models.Interface` instance with two IP addresses assigned
          through the `ip_addresses` many-to-many relationship as `database_object`
        - an InterfaceModel class like the following `NautobotInterface` as `diffsync_model`

        ```python
        class IPAddressDict(TypedDict):
            host: str
            prefix_length: int


        class NautobotInterface(NautobotModel):
            _model = Interface
            _modelname = "interface"
            _identifiers = (
                "name",
                "device__name",
            )
            _attributes = ("ip_addresses",)

            name: str
            device__name: str
            ip_addresses: List[IPAddressDict] = []
        ```

        - a field name like `ip_addresses` as the `parameter_name`

        Example return list within the above input example:

        ```python
        [
          {"host": "192.0.2.0/25", "prefix_length": 25},
          {"host": "192.0.2.128/25", "prefix_length": 25},
        ]
        ```
        """
        # Introspect type annotations to deduce which fields are of interest
        # for this many-to-many relationship.
        inner_type = get_inner_type(diffsync_model, parameter_name)
        related_objects_list = []
        # TODO: Allow for filtering, i.e. not taking into account all the objects behind the relationship.
        for related_object in getattr(database_object, parameter_name).all():
            dictionary_representation = load_typed_dict(inner_type, related_object)
            # Only use those where there is a single field defined, all 'None's will not help us.
            if any(dictionary_representation.values()):
                related_objects_list.append(dictionary_representation)
        return related_objects_list

    def _handle_custom_relationship_foreign_key(
        self, database_object, parameter_name: str, annotation: CustomRelationshipAnnotation
    ):
        """Handle a single custom relationship foreign key field."""
        relationship_associations, association_count = get_custom_relationship_associations(
            relationship=self.cache.get_from_orm(Relationship, {"label": annotation.name}),
            db_obj=database_object,
            relationship_side=annotation.side,
        )

        if association_count == 0:
            return None
        if association_count > 1:
            self.job.logger.warning(
                f"Foreign key ({database_object.__name__}.{parameter_name}) "
                "custom relationship matched two associations - this shouldn't happen."
            )

        return orm_attribute_lookup(
            getattr(
                relationship_associations.first(),
                "source" if annotation.side == RelationshipSideEnum.DESTINATION else "destination",
            ),
            # Discard the first part of the paramater name as it references the initial related object
            re.sub("^(.*?)__", "", parameter_name),
        )

    def get_or_create_metadatatype(self):
        """Retrieve or create a MetadataType object to track the last sync time of this SSoT job."""
        # MetadataType name will be extracted from the Data Source name.
        metadata_type__name = self.job.__class__.Meta.data_source

        # Create a MetadataType of type datetime
        metadata_type, _ = MetadataType.objects.get_or_create(
            name=f"Last sync from {metadata_type__name}",
            defaults={
                "data_type": "datetime",
                "description": f"Timestamp of the last sync from the Data source {metadata_type__name}",
            },
        )

        # Get All diffsync models from adapter's top_level attribute
        diffsync_models: List[BaseNautobotModel] = []
        for model_name in self.top_level:
            diffsync_model = self._get_diffsync_class(model_name)
            diffsync_models.append(diffsync_model)
            for children_parameter, _ in diffsync_model._children.items():
                diffsync_model_child = self._get_diffsync_class(model_name=children_parameter)
                diffsync_models.append(diffsync_model_child)

        for diffsync_model in diffsync_models:
            # Get nautobot model from diffsync model
            nautobot_model = diffsync_model._model
            # Attach ContentTypes to MetadataType
            content_type = ContentType.objects.get_for_model(nautobot_model)
            if content_type not in metadata_type.content_types.all():
                metadata_type.content_types.add(content_type)
            # Define scope fields per model
            obj_metadata_scope_fields = list(
                {parameter.split("__", maxsplit=1)[0] for parameter in diffsync_model.get_synced_attributes()}
            )
            self.metadata_scope_fields[diffsync_model] = obj_metadata_scope_fields

        # Define the metadata type on the adapter so that can be used on the models crud operations
        self.metadata_type = metadata_type

__init__(*args, job, sync=None, **kwargs)

Instantiate this class, but do not load data immediately from the local system.

Source code in nautobot_ssot/contrib/adapter.py

def __init__(self, *args, job, sync=None, **kwargs):
    """Instantiate this class, but do not load data immediately from the local system."""
    super().__init__(*args, **kwargs)
    self.job = job
    self.sync = sync
    self.cache: ORMCache = kwargs.pop("cache", ORMCache())
    self.metadata_type = None
    self.metadata_scope_fields = {}

get_from_orm_cache(parameters, model_class)

Retrieve an object from the ORM or the cache.

Source code in nautobot_ssot/contrib/adapter.py

def get_from_orm_cache(self, parameters: Dict, model_class: Type[Model]):
    """Retrieve an object from the ORM or the cache."""
    return self.cache.get_from_orm(model_class, parameters)

get_or_create_metadatatype()

Retrieve or create a MetadataType object to track the last sync time of this SSoT job.

Source code in nautobot_ssot/contrib/adapter.py

def get_or_create_metadatatype(self):
    """Retrieve or create a MetadataType object to track the last sync time of this SSoT job."""
    # MetadataType name will be extracted from the Data Source name.
    metadata_type__name = self.job.__class__.Meta.data_source

    # Create a MetadataType of type datetime
    metadata_type, _ = MetadataType.objects.get_or_create(
        name=f"Last sync from {metadata_type__name}",
        defaults={
            "data_type": "datetime",
            "description": f"Timestamp of the last sync from the Data source {metadata_type__name}",
        },
    )

    # Get All diffsync models from adapter's top_level attribute
    diffsync_models: List[BaseNautobotModel] = []
    for model_name in self.top_level:
        diffsync_model = self._get_diffsync_class(model_name)
        diffsync_models.append(diffsync_model)
        for children_parameter, _ in diffsync_model._children.items():
            diffsync_model_child = self._get_diffsync_class(model_name=children_parameter)
            diffsync_models.append(diffsync_model_child)

    for diffsync_model in diffsync_models:
        # Get nautobot model from diffsync model
        nautobot_model = diffsync_model._model
        # Attach ContentTypes to MetadataType
        content_type = ContentType.objects.get_for_model(nautobot_model)
        if content_type not in metadata_type.content_types.all():
            metadata_type.content_types.add(content_type)
        # Define scope fields per model
        obj_metadata_scope_fields = list(
            {parameter.split("__", maxsplit=1)[0] for parameter in diffsync_model.get_synced_attributes()}
        )
        self.metadata_scope_fields[diffsync_model] = obj_metadata_scope_fields

    # Define the metadata type on the adapter so that can be used on the models crud operations
    self.metadata_type = metadata_type

invalidate_cache(zero_out_hits=True)

Deprecated, kept for backwards compatibility.

Source code in nautobot_ssot/contrib/adapter.py

def invalidate_cache(self, zero_out_hits=True):
    """Deprecated, kept for backwards compatibility."""
    self.job.logger.warning(
        "Adapter class method `self.invalidate_cache()` is deprecated and will be removed in a future version. "
        "Use `self.cache.invalidate_cache()` instead."
    )
    self.cache.invalidate_cache(zero_out_hits=zero_out_hits)

load()

Generic implementation of the load function.

Source code in nautobot_ssot/contrib/adapter.py

def load(self):
    """Generic implementation of the load function."""
    if not hasattr(self, "top_level") or not self.top_level:
        raise ValueError("'top_level' needs to be set on the class.")

    for model_name in self.top_level:
        diffsync_model = self._get_diffsync_class(model_name)

        # This function directly mutates the diffsync store, i.e. it will create and load the objects
        # for this specific model class as well as its children without returning anything.
        self._load_objects(diffsync_model)

nautobot_ssot.jobs.base.DataSyncBaseJob

Bases: Job

Common base class for data synchronization jobs.

Works mostly as per the BaseJob API, with the following changes:

• Concrete subclasses are responsible for implementing self.sync_data() (or related hooks), not self.run().
• Subclasses may optionally define any Meta field supported by Jobs, as well as the following:
• dryrun_default - defaults to True if unspecified
• data_source and data_target as labels (by default, will use the name and/or "Nautobot" as appropriate)
• data_source_icon and data_target_icon

Source code in nautobot_ssot/jobs/base.py

class DataSyncBaseJob(Job):  # pylint: disable=too-many-instance-attributes
    """Common base class for data synchronization jobs.

    Works mostly as per the BaseJob API, with the following changes:

    - Concrete subclasses are responsible for implementing `self.sync_data()` (or related hooks), **not** `self.run()`.
    - Subclasses may optionally define any Meta field supported by Jobs, as well as the following:
      - `dryrun_default` - defaults to True if unspecified
      - `data_source` and `data_target` as labels (by default, will use the `name` and/or "Nautobot" as appropriate)
      - `data_source_icon` and `data_target_icon`
    """

    dryrun = DryRunVar(
        description="Perform a dry-run, making no actual changes to Nautobot data.",
        default=True,
    )
    memory_profiling = BooleanVar(description="Perform a memory profiling analysis.", default=False)
    parallel_loading = BooleanVar(
        description="Load source and target adapters in parallel for improved performance.",
        default=True,
    )

    def load_source_adapter(self):
        """Method to instantiate and load the SOURCE adapter into `self.source_adapter`.

        Relevant available instance attributes include:

        - self.job_result (as per Job API)
        """
        raise NotImplementedError

    def load_target_adapter(self):
        """Method to instantiate and load the TARGET adapter into `self.target_adapter`.

        Relevant available instance attributes include:

        - self.job_result (as per Job API)
        """
        raise NotImplementedError

    def calculate_diff(self):
        """Method to calculate the difference from SOURCE to TARGET adapter and store in `self.diff`.

        This is a generic implementation that you could overwrite completely in your custom logic.
        """
        if self.source_adapter is not None and self.target_adapter is not None:
            self.diff = self.source_adapter.diff_to(self.target_adapter, flags=self.diffsync_flags)
            self.sync.diff = {}
            self.sync.summary = self.diff.summary()
            self.sync.save()
            try:
                self.sync.diff = self.diff.dict()
                self.sync.save()
            except OperationalError:
                self.logger.warning("Unable to save JSON diff to the database; likely the diff is too large.")
                self.sync.refresh_from_db()
            self.logger.info(self.diff.summary())
        else:
            self.logger.warning("Not both adapters were properly initialized prior to diff calculation.")

    def execute_sync(self):
        """Method to synchronize the difference from `self.diff`, from SOURCE to TARGET adapter.

        This is a generic implementation that you could overwrite completely in your custom logic.
        """
        if self.source_adapter is not None and self.target_adapter is not None:
            self.source_adapter.sync_to(self.target_adapter, flags=self.diffsync_flags)
        else:
            self.logger.warning("Not both adapters were properly initialized prior to synchronization.")

    def _load_adapter_parallel(self, adapter_type):
        """Load an adapter in a separate thread using ThreadedAdapterLoader.

        Args:
            adapter_type: Either "source" or "target" to specify which adapter to load.

        Returns:
            tuple: (adapter_type, adapter, error, log_records, duration)
        """
        loader = ThreadedAdapterLoader(
            adapter=adapter_type,
            job=self,
            job_result=self.job_result,
            log_level=logging.DEBUG,
        )

        try:
            adapter = loader.load()
            return (adapter_type, adapter, None, loader.log_handler.records, loader.duration)
        except Exception as error:  # pylint: disable=broad-except
            return (adapter_type, None, error, loader.log_handler.records, loader.duration)

    def _load_source_adapter_parallel(self):
        """Load source adapter in a separate thread.

        Returns:
            tuple: (adapter_type, adapter, error, log_records, duration)
        """
        return self._load_adapter_parallel("source")

    def _load_target_adapter_parallel(self):
        """Load target adapter in a separate thread.

        Returns:
            tuple: (adapter_type, adapter, error, log_records, duration)
        """
        return self._load_adapter_parallel("target")

    def _load_adapters_parallel(self):  # pylint: disable=too-many-locals,too-many-branches,too-many-statements
        """Load source and target adapters in parallel using ThreadPoolExecutor.

        Returns:
            tuple: (source_adapter, target_adapter, source_duration, target_duration) after both have been loaded
        """
        source_adapter = None
        target_adapter = None
        source_error = None
        target_error = None

        # Store reference needed for logger setup in threads
        job_result = self.job_result

        source_log_records = []
        target_log_records = []
        source_duration = None
        target_duration = None

        with ThreadPoolExecutor(max_workers=2) as executor:
            # Submit both adapter loading tasks
            future_to_adapter = {
                executor.submit(self._load_source_adapter_parallel): "source",
                executor.submit(self._load_target_adapter_parallel): "target",
            }

            # Wait for both to complete
            for future in as_completed(future_to_adapter):
                adapter_type, adapter, error, log_records, duration = future.result()
                if error:
                    if adapter_type == "source":
                        source_error = error
                        source_log_records = log_records
                        source_duration = duration
                    else:
                        target_error = error
                        target_log_records = log_records
                        target_duration = duration
                else:
                    if adapter_type == "source":
                        source_adapter = adapter
                        self.source_adapter = adapter  # Ensure it's set on self
                        source_log_records = log_records
                        source_duration = duration
                    else:
                        target_adapter = adapter
                        self.target_adapter = adapter  # Ensure it's set on self
                        target_log_records = log_records
                        target_duration = duration

        # Create JobLogEntry objects for all captured log messages
        # This ensures logs from threads are visible in the Job Results
        # Merge source and target logs by timestamp to show interleaved execution

        def create_job_log_entry(record, adapter_type):
            """Create a single JobLogEntry object from a log record."""
            # Map logging levels to appropriate log levels for JobLogEntry
            if record.levelno >= logging.ERROR:
                log_level = "error"
            elif record.levelno >= logging.WARNING:
                log_level = "warning"
            elif record.levelno >= logging.INFO:
                log_level = "info"
            else:
                log_level = "debug"

            # Format the message
            message = record.getMessage()
            if record.exc_info:
                # Include exception information if present
                message += "\n" + "".join(traceback.format_exception(*record.exc_info))

            # Create JobLogEntry object with grouping set to adapter type
            # Note: JobLogEntry fields may vary by Nautobot version, using common fields
            JobLogEntry.objects.create(
                job_result=job_result,
                log_level=log_level,
                message=message,
                grouping=adapter_type,  # Group by source or target adapter
            )

        # Merge source and target logs by timestamp to show interleaved execution
        # Tag each record with its adapter type before merging
        all_log_records = []
        for record in source_log_records:
            all_log_records.append((record.created, "source", record))
        for record in target_log_records:
            all_log_records.append((record.created, "target", record))

        # Remove duplicates based on message and timestamp
        seen_messages = set()
        unique_records = []
        for timestamp, adapter_type, record in all_log_records:
            # Create a unique key for the message
            message_key = (record.getMessage(), timestamp)
            if message_key not in seen_messages:
                seen_messages.add(message_key)
                unique_records.append((timestamp, adapter_type, record))

        # Sort by timestamp to show chronological order (interleaved execution)
        unique_records.sort(key=lambda x: x[0])

        # Create JobLogEntry objects in chronological order
        for timestamp, adapter_type, record in unique_records:
            create_job_log_entry(record, adapter_type)

        # Log timing messages for each adapter to match sequential loading format
        # Create JobLogEntry objects explicitly to ensure they're stored in the database
        if source_adapter is not None and source_duration is not None:
            # Log using logger to match sequential loading behavior
            self.logger.info(
                "Source Load Time from %s: %s",
                source_adapter,
                source_duration,
            )
            # Create JobLogEntry explicitly for database persistence
            source_adapter_name = str(source_adapter) if source_adapter else "source adapter"
            timing_message = f"Source adapter ({source_adapter_name}) loaded in {source_duration}"
            self.logger.info(timing_message)
            JobLogEntry.objects.create(
                job_result=job_result,
                log_level="info",
                message=timing_message,
                grouping="source",
            )

        if target_adapter is not None and target_duration is not None:
            # Log using logger to match sequential loading behavior
            self.logger.info(
                "Target Load Time from %s: %s",
                target_adapter,
                target_duration,
            )
            # Create JobLogEntry explicitly for database persistence
            target_adapter_name = str(target_adapter) if target_adapter else "target adapter"
            timing_message = f"Target adapter ({target_adapter_name}) loaded in {target_duration}"
            self.logger.info(timing_message)
            JobLogEntry.objects.create(
                job_result=job_result,
                log_level="info",
                message=timing_message,
                grouping="target",
            )

        # Raise errors if any occurred
        if source_error:
            raise source_error
        if target_error:
            raise target_error

        return source_adapter, target_adapter, source_duration, target_duration

    def sync_data(self, memory_profiling):  # pylint: disable=too-many-statements,too-many-locals,too-many-branches
        """Method to load data from adapters, calculate diffs and sync (if not dry-run).

        It is composed by 4 methods:
        - self.load_source_adapter: instantiates the source adapter (self.source_adapter) and loads its data
        - self.load_target_adapter: instantiates the target adapter (self.target_adapter) and loads its data
        - self.calculate_diff: generates the diff from source to target adapter and stores it in self.diff
        - self.execute_sync: if not dry-run, uses the self.diff to synchronize from source to target

        This is a generic implementation that you could overwrite completely in you custom logic.
        Available instance attributes include:

        - self.sync       (Sync instance tracking this job execution)
        - self.job_result (as per Job API)
        """

        def format_size(size):  # pylint: disable=inconsistent-return-statements
            """Format a size in bytes to a human-readable string. Borrowed from stdlib tracemalloc."""
            for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
                if abs(size) < 100 and unit != "B":
                    # 3 digits (xx.x UNIT)
                    return "%.1f %s" % (  # pylint: disable=consider-using-f-string
                        size,
                        unit,
                    )
                if abs(size) < 10 * 1024 or unit == "TiB":
                    # 4 or 5 digits (xxxx UNIT)
                    return "%.0f %s" % (  # pylint: disable=consider-using-f-string
                        size,
                        unit,
                    )
                size /= 1024

        def record_memory_trace(step: str):
            """Helper function to record memory usage and reset tracemalloc stats."""
            memory_final, memory_peak = tracemalloc.get_traced_memory()
            setattr(self.sync, f"{step}_memory_final", memory_final)
            setattr(self.sync, f"{step}_memory_peak", memory_peak)
            self.sync.save()
            self.logger.info(
                "Traced memory for %s (Final, Peak): %s, %s",
                step,
                format_size(memory_final),
                format_size(memory_peak),
            )
            tracemalloc.clear_traces()

        if not self.sync:
            return

        if memory_profiling:
            tracemalloc.start()

        start_time = datetime.now()

        # Check if parallel loading is enabled (default True, can be overridden per job)
        use_parallel = getattr(self, "parallel_loading", True)
        if isinstance(use_parallel, bool):
            parallel_enabled = use_parallel
        else:
            # If it's a BooleanVar instance, get its value from kwargs or default
            parallel_enabled = getattr(self, "_parallel_loading_value", True)

        # Initialize variables for timing
        adapter_load_end_time = None
        load_target_adapter_time = None

        if parallel_enabled:
            self.logger.info("Loading source and target adapters in parallel...")
            try:
                _, _, source_duration, target_duration = self._load_adapters_parallel()
                # In parallel mode, both adapters run concurrently, so the total time
                # is the maximum of the two individual durations
                # Store the same value (total parallel time) for both to match test expectations
                if source_duration is not None and target_duration is not None:
                    parallel_duration = max(source_duration, target_duration)
                    self.sync.source_load_time = parallel_duration
                    self.sync.target_load_time = parallel_duration
                elif source_duration is not None:
                    self.sync.source_load_time = source_duration
                    self.sync.target_load_time = source_duration
                elif target_duration is not None:
                    self.sync.source_load_time = target_duration
                    self.sync.target_load_time = target_duration
                self.sync.save()
                if memory_profiling:
                    # Record memory after both adapters are loaded
                    record_memory_trace("parallel_load")
            except Exception as error:
                self.logger.error("Error during parallel adapter loading: %s", error)
                raise
        else:
            # Sequential loading (original behavior)
            self.logger.info("Loading current data from source adapter...")
            self.load_source_adapter()
            load_source_adapter_time = datetime.now()
            self.sync.source_load_time = load_source_adapter_time - start_time
            self.sync.save()
            self.logger.info(
                "Source Load Time from %s: %s",
                self.source_adapter,
                self.sync.source_load_time,
            )
            if memory_profiling:
                record_memory_trace("source_load")

            self.logger.info("Loading current data from target adapter...")
            self.load_target_adapter()
            load_target_adapter_time = datetime.now()
            adapter_load_end_time = load_target_adapter_time
            self.sync.target_load_time = load_target_adapter_time - load_source_adapter_time
            self.sync.save()
            self.logger.info(
                "Target Load Time from %s: %s",
                self.target_adapter,
                self.sync.target_load_time,
            )
            if memory_profiling:
                record_memory_trace("target_load")

        # Check if the adapter is an instance of NautobotAdapter to determine if it's a contrib implementation,
        # in which case we should create the required MetadataType object.
        if self.__class__.__name__ in settings.PLUGINS_CONFIG.get("nautobot_ssot", {}).get(
            "enable_metadata_for", []
        ) and isinstance(self.target_adapter, NautobotAdapter):
            self.target_adapter.get_or_create_metadatatype()

        # Sorting relationships must be done before calculating diffs.
        # NOTE: Disabled for the time being due to ongoing issues.
        # sort_relationships(self.source_adapter, self.target_adapter)

        # Calculate the time after adapter loading for diff calculation
        if adapter_load_end_time is None:
            adapter_load_end_time = datetime.now()

        self.logger.info("Calculating diffs...")
        self.calculate_diff()
        calculate_diff_time = datetime.now()
        self.sync.diff_time = calculate_diff_time - adapter_load_end_time
        self.sync.save()
        self.logger.info("Diff Calculation Time: %s", self.sync.diff_time)
        if memory_profiling:
            record_memory_trace("diff")

        if self.sync.dry_run:
            self.logger.info("As `dryrun` is set, skipping the actual data sync.")
        else:
            self.logger.info("Syncing from %s to %s...", self.source_adapter, self.target_adapter)
            self.execute_sync()
            execute_sync_time = datetime.now()
            self.sync.sync_time = execute_sync_time - calculate_diff_time
            self.sync.save()
            self.logger.info("Sync complete")
            self.logger.info("Sync Time: %s", self.sync.sync_time)
            if memory_profiling:
                record_memory_trace("sync")

    def lookup_object(  # pylint: disable=unused-argument
        self,
        model_name,
        unique_id,
    ) -> Optional[BaseModel]:
        """Look up the Nautobot record, if any, identified by the args.

        Optional helper method used to build more detailed/accurate SyncLogEntry records from DiffSync logs.

        Args:
            model_name (str): DiffSyncModel class name or similar class/model label.
            unique_id (str): DiffSyncModel unique_id or similar unique identifier.

        Returns:
            Optional[BaseModel]: Nautobot model instance, or None
        """
        return None

    @classmethod
    def data_mappings(cls) -> Iterable[DataMapping]:
        """List the data mappings involved in this sync job."""
        return []

    @classmethod
    def config_information(cls):
        """Return a dict of user-facing configuration information {property: value}.

        Note that this will be rendered 'as-is' in the UI, so as a general practice this
        should NOT include sensitive information such as passwords!
        """
        return {}

    def sync_log(  # pylint: disable=too-many-arguments
        self,
        action,
        status,
        message="",
        diff=None,
        synced_object=None,
        object_repr="",
    ):
        """Log a action message as a SyncLogEntry."""
        if synced_object and not object_repr:
            object_repr = repr(synced_object)

        SyncLogEntry.objects.create(
            sync=self.sync,
            action=action,
            status=status,
            message=message,
            diff=diff,
            synced_object=synced_object,
            object_repr=object_repr,
        )

    def _structlog_to_sync_log_entry(self, _logger, _log_method, event_dict):
        """Capture certain structlog messages from DiffSync into the Nautobot database."""
        if all(key in event_dict for key in ("src", "dst", "action", "model", "unique_id", "diffs", "status")):
            # The DiffSync log gives us a model name (string) and unique_id (string).
            # Try to look up the actual Nautobot object that this describes.
            synced_object = self.lookup_object(  # pylint: disable=assignment-from-none
                event_dict["model"], event_dict["unique_id"]
            )
            object_repr = repr(synced_object) if synced_object else f"{event_dict['model']} {event_dict['unique_id']}"
            self.sync_log(
                action=event_dict["action"] or SyncLogEntryActionChoices.ACTION_NO_CHANGE,
                diff=event_dict["diffs"] if event_dict["action"] else None,
                status=event_dict["status"],
                message=event_dict["event"],
                synced_object=synced_object,
                object_repr=object_repr,
            )

        return event_dict

    @classmethod
    def _get_vars(cls):
        """Extend Job._get_vars to include `dryrun` variable.

        Workaround for https://github.com/netbox-community/netbox/issues/5529
        """
        got_vars = super()._get_vars()
        if hasattr(cls, "dryrun"):
            got_vars["dryrun"] = cls.dryrun

        if hasattr(cls, "memory_profiling"):
            got_vars["memory_profiling"] = cls.memory_profiling

        if hasattr(cls, "parallel_loading"):
            got_vars["parallel_loading"] = cls.parallel_loading
        return got_vars

    def __init__(self):
        """Initialize a Job."""
        super().__init__()
        self.sync = None
        self._parallel_loading_value = True
        self.diff = None
        self.source_adapter = None
        self.target_adapter = None
        # Default diffsync flags. You can overwrite them at any time.
        self.diffsync_flags = DiffSyncFlags.CONTINUE_ON_FAILURE | DiffSyncFlags.LOG_UNCHANGED_RECORDS

    @classmethod
    def as_form(cls, data=None, files=None, initial=None, approval_view=False):
        """Render this instance as a Django form for user inputs, including a "Dry run" field."""
        form = super().as_form(data=data, files=files, initial=initial, approval_view=approval_view)
        # Set the "dryrun" widget's initial value based on our Meta attribute, if any
        form.fields["dryrun"].initial = getattr(cls.Meta, "dryrun_default", True)
        return form

    @classproperty
    def data_source(cls):
        """The system or data source providing input data for this sync."""
        return getattr(cls.Meta, "data_source", cls.name)

    @classproperty
    def data_target(cls):
        """The system or data source being modified by this sync."""
        return getattr(cls.Meta, "data_target", cls.name)

    @classproperty
    def data_source_icon(cls):
        """Icon corresponding to the data_source."""
        return getattr(cls.Meta, "data_source_icon", None)

    @classproperty
    def data_target_icon(cls):
        """Icon corresponding to the data_target."""
        return getattr(cls.Meta, "data_target_icon", None)

    def run(self, *args, **kwargs):
        """Job entry point from Nautobot - do not override!"""
        self.dryrun = kwargs.get("dryrun", True)
        self.memory_profiling = kwargs.get("memory_profiling", False)
        self.parallel_loading = kwargs.get("parallel_loading", True)
        self.sync = Sync.objects.create(
            source=self.data_source,
            target=self.data_target,
            dry_run=self.dryrun,
            job_result=self.job_result,
            start_time=timezone.now(),
            diff={},
        )

        # Store parallel_loading value for use in sync_data
        # If not provided, default to True (parallel loading enabled by default)
        self._parallel_loading_value = self.parallel_loading

        # Add _structlog_to_sync_log_entry as a processor for structlog calls from DiffSync
        structlog.configure(
            processors=[
                self._structlog_to_sync_log_entry,
                structlog.stdlib.render_to_log_kwargs,
            ],
            context_class=dict,
            logger_factory=structlog.stdlib.LoggerFactory(),
            wrapper_class=structlog.stdlib.BoundLogger,
            cache_logger_on_first_use=True,
        )
        self.sync_data(self.memory_profiling)

__init__()

Initialize a Job.

Source code in nautobot_ssot/jobs/base.py

def __init__(self):
    """Initialize a Job."""
    super().__init__()
    self.sync = None
    self._parallel_loading_value = True
    self.diff = None
    self.source_adapter = None
    self.target_adapter = None
    # Default diffsync flags. You can overwrite them at any time.
    self.diffsync_flags = DiffSyncFlags.CONTINUE_ON_FAILURE | DiffSyncFlags.LOG_UNCHANGED_RECORDS

as_form(data=None, files=None, initial=None, approval_view=False) classmethod

Render this instance as a Django form for user inputs, including a "Dry run" field.

Source code in nautobot_ssot/jobs/base.py

@classmethod
def as_form(cls, data=None, files=None, initial=None, approval_view=False):
    """Render this instance as a Django form for user inputs, including a "Dry run" field."""
    form = super().as_form(data=data, files=files, initial=initial, approval_view=approval_view)
    # Set the "dryrun" widget's initial value based on our Meta attribute, if any
    form.fields["dryrun"].initial = getattr(cls.Meta, "dryrun_default", True)
    return form

calculate_diff()

Method to calculate the difference from SOURCE to TARGET adapter and store in self.diff.

This is a generic implementation that you could overwrite completely in your custom logic.

Source code in nautobot_ssot/jobs/base.py

def calculate_diff(self):
    """Method to calculate the difference from SOURCE to TARGET adapter and store in `self.diff`.

    This is a generic implementation that you could overwrite completely in your custom logic.
    """
    if self.source_adapter is not None and self.target_adapter is not None:
        self.diff = self.source_adapter.diff_to(self.target_adapter, flags=self.diffsync_flags)
        self.sync.diff = {}
        self.sync.summary = self.diff.summary()
        self.sync.save()
        try:
            self.sync.diff = self.diff.dict()
            self.sync.save()
        except OperationalError:
            self.logger.warning("Unable to save JSON diff to the database; likely the diff is too large.")
            self.sync.refresh_from_db()
        self.logger.info(self.diff.summary())
    else:
        self.logger.warning("Not both adapters were properly initialized prior to diff calculation.")

config_information() classmethod

Return a dict of user-facing configuration information {property: value}.

Note that this will be rendered 'as-is' in the UI, so as a general practice this should NOT include sensitive information such as passwords!

Source code in nautobot_ssot/jobs/base.py

@classmethod
def config_information(cls):
    """Return a dict of user-facing configuration information {property: value}.

    Note that this will be rendered 'as-is' in the UI, so as a general practice this
    should NOT include sensitive information such as passwords!
    """
    return {}

data_mappings() classmethod

List the data mappings involved in this sync job.

Source code in nautobot_ssot/jobs/base.py

@classmethod
def data_mappings(cls) -> Iterable[DataMapping]:
    """List the data mappings involved in this sync job."""
    return []

data_source()

The system or data source providing input data for this sync.

Source code in nautobot_ssot/jobs/base.py

@classproperty
def data_source(cls):
    """The system or data source providing input data for this sync."""
    return getattr(cls.Meta, "data_source", cls.name)

data_source_icon()

Icon corresponding to the data_source.

Source code in nautobot_ssot/jobs/base.py

@classproperty
def data_source_icon(cls):
    """Icon corresponding to the data_source."""
    return getattr(cls.Meta, "data_source_icon", None)

data_target()

The system or data source being modified by this sync.

Source code in nautobot_ssot/jobs/base.py

@classproperty
def data_target(cls):
    """The system or data source being modified by this sync."""
    return getattr(cls.Meta, "data_target", cls.name)

data_target_icon()

Icon corresponding to the data_target.

Source code in nautobot_ssot/jobs/base.py

@classproperty
def data_target_icon(cls):
    """Icon corresponding to the data_target."""
    return getattr(cls.Meta, "data_target_icon", None)

execute_sync()

Method to synchronize the difference from self.diff, from SOURCE to TARGET adapter.

This is a generic implementation that you could overwrite completely in your custom logic.

Source code in nautobot_ssot/jobs/base.py

def execute_sync(self):
    """Method to synchronize the difference from `self.diff`, from SOURCE to TARGET adapter.

    This is a generic implementation that you could overwrite completely in your custom logic.
    """
    if self.source_adapter is not None and self.target_adapter is not None:
        self.source_adapter.sync_to(self.target_adapter, flags=self.diffsync_flags)
    else:
        self.logger.warning("Not both adapters were properly initialized prior to synchronization.")

load_source_adapter()

Method to instantiate and load the SOURCE adapter into self.source_adapter.

Relevant available instance attributes include:

• self.job_result (as per Job API)

Source code in nautobot_ssot/jobs/base.py

def load_source_adapter(self):
    """Method to instantiate and load the SOURCE adapter into `self.source_adapter`.

    Relevant available instance attributes include:

    - self.job_result (as per Job API)
    """
    raise NotImplementedError

load_target_adapter()

Method to instantiate and load the TARGET adapter into self.target_adapter.

Relevant available instance attributes include:

• self.job_result (as per Job API)

Source code in nautobot_ssot/jobs/base.py

def load_target_adapter(self):
    """Method to instantiate and load the TARGET adapter into `self.target_adapter`.

    Relevant available instance attributes include:

    - self.job_result (as per Job API)
    """
    raise NotImplementedError

lookup_object(model_name, unique_id)

Look up the Nautobot record, if any, identified by the args.

Optional helper method used to build more detailed/accurate SyncLogEntry records from DiffSync logs.

Parameters:

Name	Type	Description	Default
model_name	str	DiffSyncModel class name or similar class/model label.	required
unique_id	str	DiffSyncModel unique_id or similar unique identifier.	required

Returns:

Type	Description
Optional[BaseModel]	Optional[BaseModel]: Nautobot model instance, or None

Source code in nautobot_ssot/jobs/base.py

def lookup_object(  # pylint: disable=unused-argument
    self,
    model_name,
    unique_id,
) -> Optional[BaseModel]:
    """Look up the Nautobot record, if any, identified by the args.

    Optional helper method used to build more detailed/accurate SyncLogEntry records from DiffSync logs.

    Args:
        model_name (str): DiffSyncModel class name or similar class/model label.
        unique_id (str): DiffSyncModel unique_id or similar unique identifier.

    Returns:
        Optional[BaseModel]: Nautobot model instance, or None
    """
    return None

run(*args, **kwargs)

Job entry point from Nautobot - do not override!

Source code in nautobot_ssot/jobs/base.py

def run(self, *args, **kwargs):
    """Job entry point from Nautobot - do not override!"""
    self.dryrun = kwargs.get("dryrun", True)
    self.memory_profiling = kwargs.get("memory_profiling", False)
    self.parallel_loading = kwargs.get("parallel_loading", True)
    self.sync = Sync.objects.create(
        source=self.data_source,
        target=self.data_target,
        dry_run=self.dryrun,
        job_result=self.job_result,
        start_time=timezone.now(),
        diff={},
    )

    # Store parallel_loading value for use in sync_data
    # If not provided, default to True (parallel loading enabled by default)
    self._parallel_loading_value = self.parallel_loading

    # Add _structlog_to_sync_log_entry as a processor for structlog calls from DiffSync
    structlog.configure(
        processors=[
            self._structlog_to_sync_log_entry,
            structlog.stdlib.render_to_log_kwargs,
        ],
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )
    self.sync_data(self.memory_profiling)

sync_data(memory_profiling)

Method to load data from adapters, calculate diffs and sync (if not dry-run).

It is composed by 4 methods: - self.load_source_adapter: instantiates the source adapter (self.source_adapter) and loads its data - self.load_target_adapter: instantiates the target adapter (self.target_adapter) and loads its data - self.calculate_diff: generates the diff from source to target adapter and stores it in self.diff - self.execute_sync: if not dry-run, uses the self.diff to synchronize from source to target

This is a generic implementation that you could overwrite completely in you custom logic. Available instance attributes include:

• self.sync (Sync instance tracking this job execution)
• self.job_result (as per Job API)

Source code in nautobot_ssot/jobs/base.py

def sync_data(self, memory_profiling):  # pylint: disable=too-many-statements,too-many-locals,too-many-branches
    """Method to load data from adapters, calculate diffs and sync (if not dry-run).

    It is composed by 4 methods:
    - self.load_source_adapter: instantiates the source adapter (self.source_adapter) and loads its data
    - self.load_target_adapter: instantiates the target adapter (self.target_adapter) and loads its data
    - self.calculate_diff: generates the diff from source to target adapter and stores it in self.diff
    - self.execute_sync: if not dry-run, uses the self.diff to synchronize from source to target

    This is a generic implementation that you could overwrite completely in you custom logic.
    Available instance attributes include:

    - self.sync       (Sync instance tracking this job execution)
    - self.job_result (as per Job API)
    """

    def format_size(size):  # pylint: disable=inconsistent-return-statements
        """Format a size in bytes to a human-readable string. Borrowed from stdlib tracemalloc."""
        for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
            if abs(size) < 100 and unit != "B":
                # 3 digits (xx.x UNIT)
                return "%.1f %s" % (  # pylint: disable=consider-using-f-string
                    size,
                    unit,
                )
            if abs(size) < 10 * 1024 or unit == "TiB":
                # 4 or 5 digits (xxxx UNIT)
                return "%.0f %s" % (  # pylint: disable=consider-using-f-string
                    size,
                    unit,
                )
            size /= 1024

    def record_memory_trace(step: str):
        """Helper function to record memory usage and reset tracemalloc stats."""
        memory_final, memory_peak = tracemalloc.get_traced_memory()
        setattr(self.sync, f"{step}_memory_final", memory_final)
        setattr(self.sync, f"{step}_memory_peak", memory_peak)
        self.sync.save()
        self.logger.info(
            "Traced memory for %s (Final, Peak): %s, %s",
            step,
            format_size(memory_final),
            format_size(memory_peak),
        )
        tracemalloc.clear_traces()

    if not self.sync:
        return

    if memory_profiling:
        tracemalloc.start()

    start_time = datetime.now()

    # Check if parallel loading is enabled (default True, can be overridden per job)
    use_parallel = getattr(self, "parallel_loading", True)
    if isinstance(use_parallel, bool):
        parallel_enabled = use_parallel
    else:
        # If it's a BooleanVar instance, get its value from kwargs or default
        parallel_enabled = getattr(self, "_parallel_loading_value", True)

    # Initialize variables for timing
    adapter_load_end_time = None
    load_target_adapter_time = None

    if parallel_enabled:
        self.logger.info("Loading source and target adapters in parallel...")
        try:
            _, _, source_duration, target_duration = self._load_adapters_parallel()
            # In parallel mode, both adapters run concurrently, so the total time
            # is the maximum of the two individual durations
            # Store the same value (total parallel time) for both to match test expectations
            if source_duration is not None and target_duration is not None:
                parallel_duration = max(source_duration, target_duration)
                self.sync.source_load_time = parallel_duration
                self.sync.target_load_time = parallel_duration
            elif source_duration is not None:
                self.sync.source_load_time = source_duration
                self.sync.target_load_time = source_duration
            elif target_duration is not None:
                self.sync.source_load_time = target_duration
                self.sync.target_load_time = target_duration
            self.sync.save()
            if memory_profiling:
                # Record memory after both adapters are loaded
                record_memory_trace("parallel_load")
        except Exception as error:
            self.logger.error("Error during parallel adapter loading: %s", error)
            raise
    else:
        # Sequential loading (original behavior)
        self.logger.info("Loading current data from source adapter...")
        self.load_source_adapter()
        load_source_adapter_time = datetime.now()
        self.sync.source_load_time = load_source_adapter_time - start_time
        self.sync.save()
        self.logger.info(
            "Source Load Time from %s: %s",
            self.source_adapter,
            self.sync.source_load_time,
        )
        if memory_profiling:
            record_memory_trace("source_load")

        self.logger.info("Loading current data from target adapter...")
        self.load_target_adapter()
        load_target_adapter_time = datetime.now()
        adapter_load_end_time = load_target_adapter_time
        self.sync.target_load_time = load_target_adapter_time - load_source_adapter_time
        self.sync.save()
        self.logger.info(
            "Target Load Time from %s: %s",
            self.target_adapter,
            self.sync.target_load_time,
        )
        if memory_profiling:
            record_memory_trace("target_load")

    # Check if the adapter is an instance of NautobotAdapter to determine if it's a contrib implementation,
    # in which case we should create the required MetadataType object.
    if self.__class__.__name__ in settings.PLUGINS_CONFIG.get("nautobot_ssot", {}).get(
        "enable_metadata_for", []
    ) and isinstance(self.target_adapter, NautobotAdapter):
        self.target_adapter.get_or_create_metadatatype()

    # Sorting relationships must be done before calculating diffs.
    # NOTE: Disabled for the time being due to ongoing issues.
    # sort_relationships(self.source_adapter, self.target_adapter)

    # Calculate the time after adapter loading for diff calculation
    if adapter_load_end_time is None:
        adapter_load_end_time = datetime.now()

    self.logger.info("Calculating diffs...")
    self.calculate_diff()
    calculate_diff_time = datetime.now()
    self.sync.diff_time = calculate_diff_time - adapter_load_end_time
    self.sync.save()
    self.logger.info("Diff Calculation Time: %s", self.sync.diff_time)
    if memory_profiling:
        record_memory_trace("diff")

    if self.sync.dry_run:
        self.logger.info("As `dryrun` is set, skipping the actual data sync.")
    else:
        self.logger.info("Syncing from %s to %s...", self.source_adapter, self.target_adapter)
        self.execute_sync()
        execute_sync_time = datetime.now()
        self.sync.sync_time = execute_sync_time - calculate_diff_time
        self.sync.save()
        self.logger.info("Sync complete")
        self.logger.info("Sync Time: %s", self.sync.sync_time)
        if memory_profiling:
            record_memory_trace("sync")

sync_log(action, status, message='', diff=None, synced_object=None, object_repr='')

Log a action message as a SyncLogEntry.

Source code in nautobot_ssot/jobs/base.py

def sync_log(  # pylint: disable=too-many-arguments
    self,
    action,
    status,
    message="",
    diff=None,
    synced_object=None,
    object_repr="",
):
    """Log a action message as a SyncLogEntry."""
    if synced_object and not object_repr:
        object_repr = repr(synced_object)

    SyncLogEntry.objects.create(
        sync=self.sync,
        action=action,
        status=status,
        message=message,
        diff=diff,
        synced_object=synced_object,
        object_repr=object_repr,
    )

nautobot_ssot.jobs.DataSource

Bases: DataSyncBaseJob

Base class for Jobs that sync data from another data source to Nautobot.

Source code in nautobot_ssot/jobs/base.py

class DataSource(DataSyncBaseJob):
    """Base class for Jobs that sync data **from** another data source **to** Nautobot."""

    @classproperty
    def data_target(cls):
        """For a DataSource this is always Nautobot."""
        return "Nautobot"

    @classproperty
    def data_target_icon(cls):
        """For a DataSource this is always the Nautobot logo."""
        return static("img/nautobot_logo.png")

data_target()

For a DataSource this is always Nautobot.

Source code in nautobot_ssot/jobs/base.py

@classproperty
def data_target(cls):
    """For a DataSource this is always Nautobot."""
    return "Nautobot"

data_target_icon()

For a DataSource this is always the Nautobot logo.

Source code in nautobot_ssot/jobs/base.py

@classproperty
def data_target_icon(cls):
    """For a DataSource this is always the Nautobot logo."""
    return static("img/nautobot_logo.png")

nautobot_ssot.jobs.DataTarget

Bases: DataSyncBaseJob

Base class for Jobs that sync data to another data target from Nautobot.

Source code in nautobot_ssot/jobs/base.py

class DataTarget(DataSyncBaseJob):
    """Base class for Jobs that sync data **to** another data target **from** Nautobot."""

    @classproperty
    def data_source(cls):
        """For a DataTarget this is always Nautobot."""
        return "Nautobot"

    @classproperty
    def data_source_icon(cls):
        """For a DataTarget this is always the Nautobot logo."""
        return static("img/nautobot_logo.png")

data_source()

For a DataTarget this is always Nautobot.

Source code in nautobot_ssot/jobs/base.py

@classproperty
def data_source(cls):
    """For a DataTarget this is always Nautobot."""
    return "Nautobot"

data_source_icon()

For a DataTarget this is always the Nautobot logo.

Source code in nautobot_ssot/jobs/base.py

@classproperty
def data_source_icon(cls):
    """For a DataTarget this is always the Nautobot logo."""
    return static("img/nautobot_logo.png")