Single Source of Truth API Package

nautobot_ssot.models

Django Models for recording the status and progress of data synchronization between data sources.

The interaction between these models and Nautobot's native JobResult model deserves some examination.

• A JobResult is created each time a data sync is requested.
• This stores a reference to the specific sync operation requested (JobResult.name), much as a Job-related JobResult would reference the name of the Job.
• This stores a 'job_id', which this app uses to reference the specific sync instance.
• This stores the 'created' and 'completed' timestamps, and the requesting user (if any)
• This stores the overall 'status' of the job (pending, running, completed, failed, errored.)
• This stores a 'data' field which, in theory can store arbitrary JSON data, but in practice expects a fairly strict structure for logging of various status messages. This field is therefore not suitable for storage of in-depth data synchronization log messages, which have a different set of content requirements, but is used for high-level status reporting.

JobResult 1<->1 Sync 1-->n SyncLogEntry

AutomationGatewayModel

Bases: PrimaryModel

Automation Gateway model for Nautobot Itential app.

Source code in nautobot_ssot/integrations/itential/models.py

class AutomationGatewayModel(PrimaryModel):  # pylint: disable=too-many-ancestors
    """Automation Gateway model for Nautobot Itential app."""

    name = models.CharField(max_length=255, unique=True)
    description = models.CharField(max_length=512, blank=True)
    location = models.ForeignKey(
        Location,
        on_delete=models.CASCADE,
        verbose_name="Location",
        help_text="Automation Gateway manages devices from this location.",
    )
    location_descendants = models.BooleanField(
        default=True,
        verbose_name="Include Location Descendants",
        help_text="Include descendant locations.",
    )
    gateway = models.OneToOneField(
        ExternalIntegration,
        on_delete=models.CASCADE,
        verbose_name="Automation Gateway",
        help_text="Automation Gateway server defined from external integration model.",
    )
    enabled = models.BooleanField(
        default=False,
        verbose_name="Automation Gateway enabled",
        help_text="Enable or Disable the Automation Gateway from being managed by Nautobot.",
    )

    class Meta:
        """Meta class."""

        ordering = ["name", "location"]
        verbose_name = "Automation Gateway Management"
        verbose_name_plural = "Automation Gateway Management"

    def __str__(self):
        """Stringify instance."""
        return self.name

Meta

Meta class.

Source code in nautobot_ssot/integrations/itential/models.py

class Meta:
    """Meta class."""

    ordering = ["name", "location"]
    verbose_name = "Automation Gateway Management"
    verbose_name_plural = "Automation Gateway Management"

__str__()

Stringify instance.

Source code in nautobot_ssot/integrations/itential/models.py

def __str__(self):
    """Stringify instance."""
    return self.name

DiffJSONEncoder

Bases: DjangoJSONEncoder

Custom JSON encoder for the Sync.diff field.

Source code in nautobot_ssot/models.py

class DiffJSONEncoder(DjangoJSONEncoder):
    """Custom JSON encoder for the Sync.diff field."""

    def default(self, o):
        """Custom JSON encoder for the Sync.diff field."""
        if isinstance(o, set):
            return self.encode(list(o))
        return super().default(o)

default(o)

Custom JSON encoder for the Sync.diff field.

Source code in nautobot_ssot/models.py

def default(self, o):
    """Custom JSON encoder for the Sync.diff field."""
    if isinstance(o, set):
        return self.encode(list(o))
    return super().default(o)

SSOTConfig

Bases: Model

Non-db model providing user permission constraints.

Source code in nautobot_ssot/models.py

class SSOTConfig(models.Model):  # pylint: disable=nb-incorrect-base-class
    """Non-db model providing user permission constraints."""

    class Meta:
        managed = False
        default_permissions = ("view",)

SSOTInfobloxConfig

Bases: PrimaryModel

SSOT Infoblox Configuration model.

Source code in nautobot_ssot/integrations/infoblox/models.py

class SSOTInfobloxConfig(PrimaryModel):  # pylint: disable=too-many-ancestors
    """SSOT Infoblox Configuration model."""

    name = models.CharField(max_length=CHARFIELD_MAX_LENGTH, unique=True)
    # TODO: Claify how saved views can be done for child apps
    is_saved_view_model = False
    description = models.CharField(
        max_length=CHARFIELD_MAX_LENGTH,
        blank=True,
    )
    default_status = models.ForeignKey(
        to="extras.Status",
        on_delete=models.PROTECT,
        verbose_name="Default Object Status",
        help_text="Default Object Status",
    )
    infoblox_instance = models.ForeignKey(
        to="extras.ExternalIntegration",
        on_delete=models.PROTECT,
        verbose_name="Infoblox Instance Config",
        help_text="Infoblox Instance",
    )
    infoblox_wapi_version = models.CharField(
        max_length=CHARFIELD_MAX_LENGTH,
        default="v2.12",
        verbose_name="Infoblox WAPI version",
    )
    enable_sync_to_infoblox = models.BooleanField(
        default=False, verbose_name="Sync to Infoblox", help_text="Enable syncing of data from Nautobot to Infoblox."
    )
    enable_sync_to_nautobot = models.BooleanField(
        default=True, verbose_name="Sync to Nautobot", help_text="Enable syncing of data from Infoblox to Nautobot."
    )
    import_ip_addresses = models.BooleanField(
        default=False,
        verbose_name="Import IP Addresses",
    )
    import_subnets = models.BooleanField(
        default=False,
        verbose_name="Import Networks",
    )
    import_vlan_views = models.BooleanField(
        default=False,
        verbose_name="Import VLAN Views",
    )
    import_vlans = models.BooleanField(
        default=False,
        verbose_name="Import VLANs",
    )
    infoblox_sync_filters = models.JSONField(default=_get_default_sync_filters, encoder=DjangoJSONEncoder)
    infoblox_network_view_to_namespace_map = models.JSONField(
        default=_get_network_view_to_namespace_map, encoder=DjangoJSONEncoder
    )
    infoblox_dns_view_mapping = models.JSONField(default=dict, encoder=DjangoJSONEncoder, blank=True)
    cf_fields_ignore = models.JSONField(default=_get_default_cf_fields_ignore, encoder=DjangoJSONEncoder, blank=True)
    import_ipv4 = models.BooleanField(
        default=True,
        verbose_name="Import IPv4",
    )
    import_ipv6 = models.BooleanField(
        default=False,
        verbose_name="Import IPv6",
    )
    dns_record_type = models.CharField(
        max_length=CHARFIELD_MAX_LENGTH,
        default=DNSRecordTypeChoices.HOST_RECORD,
        choices=DNSRecordTypeChoices,
        verbose_name="DBS record type",
        help_text="Choose what type of Infoblox DNS record to create for IP Addresses.",
    )
    fixed_address_type = models.CharField(
        max_length=CHARFIELD_MAX_LENGTH,
        default=FixedAddressTypeChoices.DONT_CREATE_RECORD,
        choices=FixedAddressTypeChoices,
        help_text="Choose what type of Infoblox fixed IP Address record to create.",
    )
    job_enabled = models.BooleanField(
        default=False,
        verbose_name="Enabled for Sync Job",
        help_text="Enable use of this configuration in the sync jobs.",
    )
    infoblox_deletable_models = models.JSONField(
        encoder=DjangoJSONEncoder,
        default=list,
        blank=True,
        help_text="Model types that can be deleted in Infoblox.",
    )
    nautobot_deletable_models = models.JSONField(
        encoder=DjangoJSONEncoder, default=list, blank=True, help_text="Model types that can be deleted in Nautobot."
    )

    class Meta:
        """Meta class for SSOTInfobloxConfig."""

        verbose_name = "SSOT Infoblox Config"
        verbose_name_plural = "SSOT Infoblox Configs"

    def __str__(self):
        """String representation of singleton instance."""
        return self.name

    def _clean_infoblox_sync_filters(self):  # pylint: disable=too-many-branches
        """Performs validation of the infoblox_sync_filters field."""
        allowed_keys = {"network_view", "prefixes_ipv4", "prefixes_ipv6"}

        if not isinstance(self.infoblox_sync_filters, list):
            raise ValidationError({"infoblox_sync_filters": "Sync filters must be a list."})

        if len(self.infoblox_sync_filters) == 0:
            raise ValidationError(
                {
                    "infoblox_sync_filters": 'At least one filter must be defined. You can use the default one: [{"network_view": "default"}]'
                }
            )

        network_views = set()
        for sync_filter in self.infoblox_sync_filters:
            if not isinstance(sync_filter, dict):
                raise ValidationError({"infoblox_sync_filters": "Sync filter must be a dict."})
            invalid_keys = set(sync_filter.keys()) - allowed_keys
            if invalid_keys:
                raise ValidationError(
                    {"infoblox_sync_filters": f"Invalid keys found in the sync filter: {', '.join(invalid_keys)}."}
                )

            if "network_view" not in sync_filter:
                raise ValidationError({"infoblox_sync_filters": "Sync filter must have `network_view` key defined."})
            network_view = sync_filter["network_view"]
            if not isinstance(network_view, str):
                raise ValidationError({"infoblox_sync_filters": "Value of the `network_view` key must be a string."})

            if network_view in network_views:
                raise ValidationError(
                    {
                        "infoblox_sync_filters": f"Duplicate value for the `network_view` found: {sync_filter['network_view']}."
                    }
                )
            network_views.add(network_view)

            if "prefixes_ipv4" in sync_filter:
                if not isinstance(sync_filter["prefixes_ipv4"], list):
                    raise ValidationError({"infoblox_sync_filters": "Value of the `prefixes_ipv4` key must be a list."})
                if not sync_filter["prefixes_ipv4"]:
                    raise ValidationError(
                        {"infoblox_sync_filters": "Value of the `prefixes_ipv4` key must not be an empty list."}
                    )
                for prefix in sync_filter["prefixes_ipv4"]:
                    try:
                        if "/" not in prefix:
                            raise ValidationError(
                                {
                                    "infoblox_sync_filters": f"IPv4 prefix must have a prefix length defined using `/` format: {prefix}."
                                }
                            )
                        ipaddress.IPv4Network(prefix)
                    except (ValueError, TypeError) as error:
                        raise ValidationError(  # pylint: disable=raise-missing-from
                            {"infoblox_sync_filters": f"IPv4 prefix parsing error: {str(error)}."}
                        )

            if "prefixes_ipv6" in sync_filter:
                if not isinstance(sync_filter["prefixes_ipv6"], list):
                    raise ValidationError({"infoblox_sync_filters": "Value of the `prefixes_ipv6` key must be a list."})
                if not sync_filter["prefixes_ipv6"]:
                    raise ValidationError(
                        {"infoblox_sync_filters": "Value of the `prefixes_ipv6` key must not be an empty list."}
                    )
                for prefix in sync_filter["prefixes_ipv6"]:
                    try:
                        if "/" not in prefix:
                            raise ValidationError(
                                {
                                    "infoblox_sync_filters": f"IPv6 prefix must have a prefix length defined using `/` format: {prefix}."
                                }
                            )
                        ipaddress.IPv6Network(prefix)
                    except (ValueError, TypeError) as error:
                        raise ValidationError(  # pylint: disable=raise-missing-from
                            {"infoblox_sync_filters": f"IPv6 prefix parsing error: {str(error)}."}
                        )

    def _clean_infoblox_network_view_to_namespace_map(self):  # pylint: disable=too-many-branches
        """Performs validation of the infoblox_network_view_to_namespace_map field."""
        if not isinstance(self.infoblox_network_view_to_namespace_map, dict):
            raise ValidationError(
                {"infoblox_network_view_to_namespace_map": "Namespace/View mappings must be a dictionary."}
            )

    def _clean_infoblox_instance(self):
        """Performs validation of the infoblox_instance field."""
        if not self.infoblox_instance.secrets_group:
            raise ValidationError({"infoblox_instance": "Infoblox instance must have Secrets groups assigned."})
        try:
            self.infoblox_instance.secrets_group.get_secret_value(
                access_type=SecretsGroupAccessTypeChoices.TYPE_REST,
                secret_type=SecretsGroupSecretTypeChoices.TYPE_USERNAME,
            )
        except SecretsGroupAssociation.DoesNotExist:
            raise ValidationError(  # pylint: disable=raise-missing-from
                {
                    "infoblox_instance": "Secrets group for the Infoblox instance must have secret with type Username and access type REST."
                }
            )
        try:
            self.infoblox_instance.secrets_group.get_secret_value(
                access_type=SecretsGroupAccessTypeChoices.TYPE_REST,
                secret_type=SecretsGroupSecretTypeChoices.TYPE_PASSWORD,
            )
        except SecretsGroupAssociation.DoesNotExist:
            raise ValidationError(  # pylint: disable=raise-missing-from
                {
                    "infoblox_instance": "Secrets group for the Infoblox instance must have secret with type Password and access type REST."
                }
            )

    def _clean_import_ip(self):
        """Performs validation of the import_ipv* fields."""
        if not (self.import_ipv4 or self.import_ipv6):
            raise ValidationError(
                {
                    "import_ipv4": "At least one of `import_ipv4` or `import_ipv6` must be set to True.",
                    "import_ipv6": "At least one of `import_ipv4` or `import_ipv6` must be set to True.",
                }
            )

    def _clean_infoblox_dns_view_mapping(self):
        """Performs validation of the infoblox_dns_view_mapping field."""
        if not isinstance(self.infoblox_dns_view_mapping, dict):
            raise ValidationError(
                {
                    "infoblox_dns_view_mapping": "`infoblox_dns_view_mapping` must be a dictionary mapping network view names to dns view names.",
                },
            )

    def _clean_cf_fields_ignore(self):
        """Performs validation of the cf_fields_ignore field."""
        if not isinstance(self.cf_fields_ignore, dict):
            raise ValidationError(
                {
                    "cf_fields_ignore": "`cf_fields_ignore` must be a dictionary.",
                },
            )
        for key, value in self.cf_fields_ignore.items():
            if key not in (
                "extensible_attributes",
                "custom_fields",
            ):
                raise ValidationError(
                    {
                        "cf_fields_ignore": f"Invalid key name `{key}`. Only `extensible_attributes` and `custom_fields` are allowed.",
                    },
                )
            if not isinstance(value, list) or {type(el) for el in value} - {str}:
                raise ValidationError(
                    {
                        "cf_fields_ignore": f"Value of key `{key}` must be a list of strings.",
                    },
                )

    def _clean_deletable_model_types(self):
        """Performs validation of infoblox_deletable_models and nautobot_deletable_models."""
        for model in self.infoblox_deletable_models:
            if model not in InfobloxDeletableModelChoices.values():
                raise ValidationError(
                    {
                        "infoblox_deletable_models": f"Model `{model}` is not a valid choice.",
                    },
                )

        for model in self.nautobot_deletable_models:
            if model not in NautobotDeletableModelChoices.values():
                raise ValidationError(
                    {
                        "nautobot_deletable_models": f"Model `{model}` is not a valid choice.",
                    },
                )

    def clean(self):
        """Clean method for SSOTInfobloxConfig."""
        super().clean()

        self._clean_infoblox_sync_filters()
        self._clean_infoblox_network_view_to_namespace_map()
        self._clean_infoblox_instance()
        self._clean_import_ip()
        self._clean_infoblox_dns_view_mapping()
        self._clean_cf_fields_ignore()
        self._clean_deletable_model_types()

Meta

Meta class for SSOTInfobloxConfig.

Source code in nautobot_ssot/integrations/infoblox/models.py

class Meta:
    """Meta class for SSOTInfobloxConfig."""

    verbose_name = "SSOT Infoblox Config"
    verbose_name_plural = "SSOT Infoblox Configs"

__str__()

String representation of singleton instance.

Source code in nautobot_ssot/integrations/infoblox/models.py

def __str__(self):
    """String representation of singleton instance."""
    return self.name

clean()

Clean method for SSOTInfobloxConfig.

Source code in nautobot_ssot/integrations/infoblox/models.py

def clean(self):
    """Clean method for SSOTInfobloxConfig."""
    super().clean()

    self._clean_infoblox_sync_filters()
    self._clean_infoblox_network_view_to_namespace_map()
    self._clean_infoblox_instance()
    self._clean_import_ip()
    self._clean_infoblox_dns_view_mapping()
    self._clean_cf_fields_ignore()
    self._clean_deletable_model_types()

SSOTServiceNowConfig

Bases: BaseModel

Singleton data model describing the configuration of this app.

Source code in nautobot_ssot/integrations/servicenow/models.py

class SSOTServiceNowConfig(BaseModel):  # pylint: disable=nb-string-field-blank-null
    """Singleton data model describing the configuration of this app."""

    def delete(self, *args, **kwargs):
        """Cannot be deleted."""

    @classmethod
    def load(cls):
        """Singleton instance getter."""
        if cls.objects.all().exists():
            return cls.objects.first()
        return cls.objects.create()

    servicenow_instance = models.CharField(
        max_length=100,
        blank=True,
        help_text="ServiceNow instance name, will be used as <code>&lt;instance&gt;.servicenow.com</code>.",
    )

    servicenow_secrets = models.ForeignKey(
        to="extras.SecretsGroup",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
        help_text="Secrets group for authentication to ServiceNow. Should contain a REST username and REST password.",
    )

    def __str__(self):
        """String representation of singleton instance."""
        return "SSoT ServiceNow Configuration"

    def get_absolute_url(self, api=False):  # pylint: disable=unused-argument
        """Get URL for the associated configuration view."""
        return reverse("plugins:nautobot_ssot:servicenow_config")

__str__()

String representation of singleton instance.

Source code in nautobot_ssot/integrations/servicenow/models.py

def __str__(self):
    """String representation of singleton instance."""
    return "SSoT ServiceNow Configuration"

delete(*args, **kwargs)

Cannot be deleted.

Source code in nautobot_ssot/integrations/servicenow/models.py

def delete(self, *args, **kwargs):
    """Cannot be deleted."""

get_absolute_url(api=False)

Get URL for the associated configuration view.

Source code in nautobot_ssot/integrations/servicenow/models.py

def get_absolute_url(self, api=False):  # pylint: disable=unused-argument
    """Get URL for the associated configuration view."""
    return reverse("plugins:nautobot_ssot:servicenow_config")

load() classmethod

Singleton instance getter.

Source code in nautobot_ssot/integrations/servicenow/models.py

@classmethod
def load(cls):
    """Singleton instance getter."""
    if cls.objects.all().exists():
        return cls.objects.first()
    return cls.objects.create()

Sync

Bases: BaseModel

High-level overview of a data sync event/process/attempt.

Essentially an extension of the JobResult model to add a few additional fields.

Source code in nautobot_ssot/models.py

@extras_features(
    "custom_links",
)
class Sync(BaseModel):  # pylint: disable=nb-string-field-blank-null
    """High-level overview of a data sync event/process/attempt.

    Essentially an extension of the JobResult model to add a few additional fields.
    """

    source = models.CharField(max_length=64, help_text="System data is read from")
    target = models.CharField(max_length=64, help_text="System data is written to")

    start_time = models.DateTimeField(blank=True, null=True, db_index=True)
    # end_time is represented by the job_result.date_done field
    source_load_time = models.DurationField(blank=True, null=True)
    target_load_time = models.DurationField(blank=True, null=True)
    diff_time = models.DurationField(blank=True, null=True)
    sync_time = models.DurationField(blank=True, null=True)
    source_load_memory_final = models.PositiveBigIntegerField(blank=True, null=True)
    source_load_memory_peak = models.PositiveBigIntegerField(blank=True, null=True)
    target_load_memory_final = models.PositiveBigIntegerField(blank=True, null=True)
    target_load_memory_peak = models.PositiveBigIntegerField(blank=True, null=True)
    diff_memory_final = models.PositiveBigIntegerField(blank=True, null=True)
    diff_memory_peak = models.PositiveBigIntegerField(blank=True, null=True)
    sync_memory_final = models.PositiveBigIntegerField(blank=True, null=True)
    sync_memory_peak = models.PositiveBigIntegerField(blank=True, null=True)

    dry_run = models.BooleanField(
        default=False,
        help_text="Report what data would be synced but do not make any changes",
    )
    diff = models.JSONField(blank=True, encoder=DiffJSONEncoder)
    summary = models.JSONField(blank=True, null=True)

    job_result = models.ForeignKey(to=JobResult, on_delete=models.CASCADE, blank=True, null=True)
    hide_in_diff_view = True

    class Meta:
        """Metaclass attributes of Sync model."""

        ordering = ["start_time"]

    def __str__(self):
        """String representation of a Sync instance."""
        return f"{self.source} → {self.target}, {date_format(self.start_time, format=settings.SHORT_DATETIME_FORMAT)}"

    def get_absolute_url(self, api=False):
        """Get the detail-view URL for this instance."""
        return reverse("plugins:nautobot_ssot:sync", kwargs={"pk": self.pk})

    @classmethod
    def annotated_queryset(cls):
        """Construct an efficient queryset for this model and related data."""
        return (
            cls.objects.defer("diff")
            .select_related("job_result")
            .prefetch_related("logs")
            .annotate(
                num_unchanged=models.Count(
                    "log",
                    filter=models.Q(log__action=SyncLogEntryActionChoices.ACTION_NO_CHANGE),
                ),
                num_created=models.Count(
                    "log",
                    filter=models.Q(log__action=SyncLogEntryActionChoices.ACTION_CREATE),
                ),
                num_updated=models.Count(
                    "log",
                    filter=models.Q(log__action=SyncLogEntryActionChoices.ACTION_UPDATE),
                ),
                num_deleted=models.Count(
                    "log",
                    filter=models.Q(log__action=SyncLogEntryActionChoices.ACTION_DELETE),
                ),
                num_succeeded=models.Count(
                    "log",
                    filter=models.Q(log__status=SyncLogEntryStatusChoices.STATUS_SUCCESS),
                ),
                num_failed=models.Count(
                    "log",
                    filter=models.Q(log__status=SyncLogEntryStatusChoices.STATUS_FAILURE),
                ),
                num_errored=models.Count(
                    "log",
                    filter=models.Q(log__status=SyncLogEntryStatusChoices.STATUS_ERROR),
                ),
            )
        )

    @property
    def duration(self):  # pylint: disable=inconsistent-return-statements
        """Total execution time of this Sync."""
        if not self.start_time:
            return timedelta()  # zero
        if not self.job_result or self.job_result.status == JobResultStatusChoices.STATUS_PENDING:
            return now() - self.start_time
        if self.job_result and self.job_result.date_done:
            return self.job_result.date_done - self.start_time

    def get_source_url(self):
        """Get the absolute url of the source worker associated with this instance."""
        if self.source == "Nautobot" or not self.job_result:
            return None
        return reverse(
            "plugins:nautobot_ssot:data_source",
            kwargs={"class_path": self.job_result.job_model.class_path},
        )

    def get_target_url(self):
        """Get the absolute url of the target worker associated with this instance."""
        if self.target == "Nautobot" or not self.job_result:
            return None
        return reverse(
            "plugins:nautobot_ssot:data_target",
            kwargs={"class_path": self.job_result.job_model.class_path},
        )

duration property

Total execution time of this Sync.

Meta

Metaclass attributes of Sync model.

Source code in nautobot_ssot/models.py

class Meta:
    """Metaclass attributes of Sync model."""

    ordering = ["start_time"]

__str__()

String representation of a Sync instance.

Source code in nautobot_ssot/models.py

def __str__(self):
    """String representation of a Sync instance."""
    return f"{self.source} → {self.target}, {date_format(self.start_time, format=settings.SHORT_DATETIME_FORMAT)}"

annotated_queryset() classmethod

Construct an efficient queryset for this model and related data.

Source code in nautobot_ssot/models.py

@classmethod
def annotated_queryset(cls):
    """Construct an efficient queryset for this model and related data."""
    return (
        cls.objects.defer("diff")
        .select_related("job_result")
        .prefetch_related("logs")
        .annotate(
            num_unchanged=models.Count(
                "log",
                filter=models.Q(log__action=SyncLogEntryActionChoices.ACTION_NO_CHANGE),
            ),
            num_created=models.Count(
                "log",
                filter=models.Q(log__action=SyncLogEntryActionChoices.ACTION_CREATE),
            ),
            num_updated=models.Count(
                "log",
                filter=models.Q(log__action=SyncLogEntryActionChoices.ACTION_UPDATE),
            ),
            num_deleted=models.Count(
                "log",
                filter=models.Q(log__action=SyncLogEntryActionChoices.ACTION_DELETE),
            ),
            num_succeeded=models.Count(
                "log",
                filter=models.Q(log__status=SyncLogEntryStatusChoices.STATUS_SUCCESS),
            ),
            num_failed=models.Count(
                "log",
                filter=models.Q(log__status=SyncLogEntryStatusChoices.STATUS_FAILURE),
            ),
            num_errored=models.Count(
                "log",
                filter=models.Q(log__status=SyncLogEntryStatusChoices.STATUS_ERROR),
            ),
        )
    )

get_absolute_url(api=False)

Get the detail-view URL for this instance.

Source code in nautobot_ssot/models.py

def get_absolute_url(self, api=False):
    """Get the detail-view URL for this instance."""
    return reverse("plugins:nautobot_ssot:sync", kwargs={"pk": self.pk})

get_source_url()

Get the absolute url of the source worker associated with this instance.

Source code in nautobot_ssot/models.py

def get_source_url(self):
    """Get the absolute url of the source worker associated with this instance."""
    if self.source == "Nautobot" or not self.job_result:
        return None
    return reverse(
        "plugins:nautobot_ssot:data_source",
        kwargs={"class_path": self.job_result.job_model.class_path},
    )

get_target_url()

Get the absolute url of the target worker associated with this instance.

Source code in nautobot_ssot/models.py

def get_target_url(self):
    """Get the absolute url of the target worker associated with this instance."""
    if self.target == "Nautobot" or not self.job_result:
        return None
    return reverse(
        "plugins:nautobot_ssot:data_target",
        kwargs={"class_path": self.job_result.job_model.class_path},
    )

SyncLogEntry

Bases: BaseModel

Record of a single event during a data sync operation.

Detailed sync logs are recorded in this model, rather than in JobResult.data, because JobResult.data imposes fairly strict expectations about the structure of its contents that do not align well with the requirements of this app. Also, storing log entries as individual database records rather than a single JSON blob allows us to filter, query, sort, etc. as desired.

This model somewhat "shadows" Nautobot's built-in ObjectChange model; the key distinction to bear in mind is that an ObjectChange reflects a change that did happen, while a SyncLogEntry may reflect this or may reflect a change that could not happen or failed. Additionally, if we're syncing data from Nautobot to a different system as data target, the data isn't changing in Nautobot, so there will be no ObjectChange record.

Source code in nautobot_ssot/models.py

class SyncLogEntry(BaseModel):  # pylint: disable=nb-string-field-blank-null
    """Record of a single event during a data sync operation.

    Detailed sync logs are recorded in this model, rather than in JobResult.data, because
    JobResult.data imposes fairly strict expectations about the structure of its contents
    that do not align well with the requirements of this app. Also, storing log entries as individual
    database records rather than a single JSON blob allows us to filter, query, sort, etc. as desired.

    This model somewhat "shadows" Nautobot's built-in ObjectChange model; the key distinction to
    bear in mind is that an ObjectChange reflects a change that *did happen*, while a SyncLogEntry
    may reflect this or may reflect a change that *could not happen* or *failed*.
    Additionally, if we're syncing data from Nautobot to a different system as data target,
    the data isn't changing in Nautobot, so there will be no ObjectChange record.
    """

    sync = models.ForeignKey(to=Sync, on_delete=models.CASCADE, related_name="logs", related_query_name="log")
    timestamp = models.DateTimeField(auto_now_add=True, db_index=True)

    action = models.CharField(max_length=32, choices=SyncLogEntryActionChoices)
    status = models.CharField(max_length=32, choices=SyncLogEntryStatusChoices)
    diff = models.JSONField(blank=True, null=True, encoder=DiffJSONEncoder)

    synced_object_type = models.ForeignKey(
        to=ContentType,
        blank=True,
        null=True,
        on_delete=models.PROTECT,
    )
    synced_object_id = models.UUIDField(blank=True, null=True)
    synced_object = GenericForeignKey(ct_field="synced_object_type", fk_field="synced_object_id")

    object_repr = models.TextField(blank=True, default="", editable=False)

    message = models.TextField(blank=True)

    hide_in_diff_view = True

    class Meta:
        """Metaclass attributes of SyncLogEntry."""

        verbose_name_plural = "sync log entries"
        ordering = ["sync", "timestamp"]

    def get_action_class(self):
        """Map self.action to a Bootstrap label class."""
        return {
            SyncLogEntryActionChoices.ACTION_NO_CHANGE: "default",
            SyncLogEntryActionChoices.ACTION_CREATE: "success",
            SyncLogEntryActionChoices.ACTION_UPDATE: "info",
            SyncLogEntryActionChoices.ACTION_DELETE: "warning",
        }.get(self.action)

    def get_status_class(self):
        """Map self.status to a Bootstrap label class."""
        return {
            SyncLogEntryStatusChoices.STATUS_SUCCESS: "success",
            SyncLogEntryStatusChoices.STATUS_FAILURE: "warning",
            SyncLogEntryStatusChoices.STATUS_ERROR: "danger",
        }.get(self.status)

Meta

Metaclass attributes of SyncLogEntry.

Source code in nautobot_ssot/models.py

class Meta:
    """Metaclass attributes of SyncLogEntry."""

    verbose_name_plural = "sync log entries"
    ordering = ["sync", "timestamp"]

get_action_class()

Map self.action to a Bootstrap label class.

Source code in nautobot_ssot/models.py

def get_action_class(self):
    """Map self.action to a Bootstrap label class."""
    return {
        SyncLogEntryActionChoices.ACTION_NO_CHANGE: "default",
        SyncLogEntryActionChoices.ACTION_CREATE: "success",
        SyncLogEntryActionChoices.ACTION_UPDATE: "info",
        SyncLogEntryActionChoices.ACTION_DELETE: "warning",
    }.get(self.action)

get_status_class()

Map self.status to a Bootstrap label class.

Source code in nautobot_ssot/models.py

def get_status_class(self):
    """Map self.status to a Bootstrap label class."""
    return {
        SyncLogEntryStatusChoices.STATUS_SUCCESS: "success",
        SyncLogEntryStatusChoices.STATUS_FAILURE: "warning",
        SyncLogEntryStatusChoices.STATUS_ERROR: "danger",
    }.get(self.status)