DRAFT: Add model attr type enum

changes/1169.added

@@ -0,0 +1,2 @@
+Added `AttributeType` enum class.
+Added class method to `NautobotModel` to get enum type of given attribute.

changes/1169.changed

@@ -0,0 +1 @@
+Changed `_handle_single_paramater` in `NautobotAdapter` to utilize `AttributeType` enum class.

changes/1169.fixed

@@ -0,0 +1,2 @@
+Fixed bug in `DiffSyncModelUtilityMixin.get_attr_args()` for Annoted type hints wrapped in `Optional[]` tag.
+Fix hashing issue on Custom Annotations.

nautobot_ssot/contrib/adapter.py

@@ -16,8 +16,8 @@
 from nautobot.extras.models.metadata import MetadataType
 
 from nautobot_ssot.contrib.base import BaseNautobotAdapter, BaseNautobotModel
+from nautobot_ssot.contrib.enums import AttributeType
 from nautobot_ssot.contrib.types import (
-    CustomFieldAnnotation,
     CustomRelationshipAnnotation,
     RelationshipSideEnum,
 )
@@ -31,8 +31,7 @@
 
 
 class NautobotAdapter(Adapter, BaseNautobotAdapter):
-    """
-    Adapter for loading data from Nautobot through the ORM.
+    """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.
     """
@@ -76,52 +75,36 @@ def _load_objects(self, diffsync_model: BaseNautobotModel):
             self._load_single_object(database_object, diffsync_model, parameter_names)
 
     def _handle_single_parameter(self, parameters, parameter_name, database_object, diffsync_model):
-        # Handle custom fields and custom relationships. See CustomFieldAnnotation and CustomRelationshipAnnotation
-        # docstrings for more details.
-        annotation = diffsync_model.get_attr_annotation(parameter_name)
-        if isinstance(annotation, CustomFieldAnnotation):
-            field_key = annotation.key or annotation.name
-            if field_key in database_object.cf:
-                parameters[parameter_name] = database_object.cf[field_key]
+        # Handle parameter overrides
+        # TODO: Overrides would be better suited in the associated Model.
+        if hasattr(self, f"load_param_{parameter_name}"):
+            parameters[parameter_name] = getattr(self, f"load_param_{parameter_name}")(parameter_name, database_object)
             return
 
-        is_custom_relationship = isinstance(annotation, CustomRelationshipAnnotation)
-
-        # 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 is_custom_relationship:
+        match diffsync_model.get_attr_enum(parameter_name):
+            case AttributeType.STANDARD:
+                parameters[parameter_name] = getattr(database_object, parameter_name)
+            case AttributeType.FOREIGN_KEY:
+                parameters[parameter_name] = orm_attribute_lookup(database_object, parameter_name)
+            case AttributeType.N_TO_MANY_RELATIONSHIP:
+                parameters[parameter_name] = self._handle_to_many_relationship(
+                    database_object, diffsync_model, parameter_name
+                )
+            case AttributeType.CUSTOM_FIELD:
+                annotation = diffsync_model.get_attr_annotation(parameter_name)
+                field_key = annotation.key or annotation.name
+                if field_key in database_object.cf:
+                    parameters[parameter_name] = database_object.cf[field_key]
+            case AttributeType.CUSTOM_FOREIGN_KEY:
                 parameters[parameter_name] = self._handle_custom_relationship_foreign_key(
-                    database_object, parameter_name, annotation
+                    database_object,
+                    parameter_name,
+                    diffsync_model.get_attr_annotation(parameter_name),
+                )
+            case AttributeType.CUSTOM_N_TO_MANY_RELATIONSHIP:
+                parameters[parameter_name] = self._handle_custom_relationship_to_many_relationship(
+                    database_object, diffsync_model, parameter_name, diffsync_model.get_attr_annotation(parameter_name)
                 )
-            else:
-                parameters[parameter_name] = orm_attribute_lookup(database_object, parameter_name)
-            return
-
-        # Handling of one- and many-to custom relationship fields:
-        if annotation:
-            parameters[parameter_name] = self._handle_custom_relationship_to_many_relationship(
-                database_object, diffsync_model, parameter_name, 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."""

nautobot_ssot/contrib/enums.py

@@ -1,6 +1,6 @@
 """Enums used in SSoT contrib processes."""
 
-from enum import Enum
+from enum import Enum, auto
 
 
 class SortType(Enum):
@@ -17,3 +17,14 @@ class RelationshipSideEnum(Enum):
 
     SOURCE = "SOURCE"
     DESTINATION = "DESTINATION"
+
+
+class AttributeType(Enum):
+    """Enum for identifying DiffSync model attribute types as used in contrib."""
+
+    STANDARD = auto()
+    FOREIGN_KEY = auto()
+    N_TO_MANY_RELATIONSHIP = auto()
+    CUSTOM_FIELD = auto()
+    CUSTOM_FOREIGN_KEY = auto()
+    CUSTOM_N_TO_MANY_RELATIONSHIP = auto()

nautobot_ssot/contrib/model.py

@@ -5,6 +5,7 @@
 
 from collections import defaultdict
 from datetime import datetime
+from functools import lru_cache
 
 from diffsync import DiffSyncModel
 from diffsync.exceptions import ObjectCrudException, ObjectNotCreated, ObjectNotDeleted, ObjectNotUpdated
@@ -16,9 +17,9 @@
 from nautobot.extras.models.metadata import ObjectMetadata
 
 from nautobot_ssot.contrib.base import BaseNautobotModel
+from nautobot_ssot.contrib.enums import AttributeType
 from nautobot_ssot.contrib.types import (
     CustomFieldAnnotation,
-    CustomRelationshipAnnotation,
     RelationshipSideEnum,
 )
 from nautobot_ssot.utils.diffsync import DiffSyncModelUtilityMixin
@@ -115,6 +116,24 @@
 
         return super().create(adapter, ids, attrs)
 
+    @classmethod
+    @lru_cache
+    def get_attr_enum(cls, attr_name: str) -> AttributeType:
+        """Return `AttributeType` enum value for type hinted attribute."""
+        annotation = cls.get_attr_annotation(attr_name)
+        if isinstance(annotation, CustomFieldAnnotation):
+            return AttributeType.CUSTOM_FIELD
+        if "__" in attr_name:
+            if annotation:
+                return AttributeType.CUSTOM_FOREIGN_KEY
+            return AttributeType.FOREIGN_KEY
+        if annotation:
+            return AttributeType.CUSTOM_N_TO_MANY_RELATIONSHIP
+        django_field = cls._model._meta.get_field(attr_name)
+        if django_field.many_to_many or django_field.one_to_many:
+            return AttributeType.N_TO_MANY_RELATIONSHIP
+        return AttributeType.STANDARD
+
     @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.
@@ -128,88 +147,63 @@
         """
         cls._check_field(field)
 
-        # Handle custom fields. See CustomFieldAnnotation docstring for more details.
         annotation = cls.get_attr_annotation(field)
-        if isinstance(annotation, CustomFieldAnnotation):
-            obj.cf[annotation.key] = value
-            return
-
-        custom_relationship_annotation = annotation if isinstance(annotation, CustomRelationshipAnnotation) else None
-
-        # 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:
+        match cls.get_attr_enum(field):
+            case AttributeType.STANDARD:
+                setattr(obj, field, value)
+            case AttributeType.FOREIGN_KEY:
+                related_model, lookup = field.split("__", maxsplit=1)
                 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)
+            case AttributeType.N_TO_MANY_RELATIONSHIP:
+                django_field = cls._model._meta.get_field(field)
+                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
+            case AttributeType.CUSTOM_FIELD:
+                obj.cf[annotation.key] = value
+            case AttributeType.CUSTOM_FOREIGN_KEY:
+                related_model, lookup = field.split("__", maxsplit=1)
+                relationship_fields["custom_relationship_foreign_keys"][related_model][lookup] = value
+                relationship_fields["custom_relationship_foreign_keys"][related_model]["_annotation"] = annotation
+            case AttributeType.CUSTOM_N_TO_MANY_RELATIONSHIP:
+                relationship = adapter.get_from_orm_cache({"label": annotation.name}, Relationship)
+                if 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 annotation.side == RelationshipSideEnum.DESTINATION
+                ):
+                    relationship_fields["custom_relationship_foreign_keys"][field] = {
+                        **value,
+                        "_annotation": annotation,
+                    }
+                else:
+                    relationship_fields["custom_relationship_many_to_many_fields"][field] = {
+                        "annotation": annotation,
+                        "objects": [
+                            adapter.get_from_orm_cache(parameters, related_model_class) for parameters in value
+                        ],
+                    }
 
     @classmethod
     def _update_obj_with_parameters(cls, obj, parameters, adapter):
         """Update a given Nautobot ORM object with the given parameters."""
+        # TODO: Use Dataclasses instead of dictionaries for structured data storage and tracking.
         relationship_fields = {
             # Example: {"group": {"name": "Group Name", "_model_class": TenantGroup}}
             "foreign_keys": defaultdict(dict),

nautobot_ssot/contrib/types.py

@@ -12,6 +12,10 @@
 class CustomAnnotation:
     """Base class used to identify custom annotations in SSoT operations."""
 
+    def __hash__(self):
+        """Return a hash of the class instance."""
+        return hash(frozenset({"class": self.__class__} | self.__dict__))
+
 
 @dataclass
 class CustomRelationshipAnnotation(CustomAnnotation):

nautobot_ssot/jobs/examples.py

@@ -530,7 +530,7 @@
         self.load_device_types()
         self.load_platforms()
         self.load_devices()
-        # self.load_interfaces()
+        self.load_interfaces()
 
     def load_location_types(self):
         """Load LocationType data from the remote Nautobot instance.

nautobot_ssot/tests/contrib/models.py

@@ -0,0 +1,62 @@
+"""Example NautobotModel instances for unittests."""
+
+from typing import Annotated, List, Optional
+
+from nautobot.dcim.models import Device, LocationType
+
+from nautobot_ssot.contrib.enums import RelationshipSideEnum
+from nautobot_ssot.contrib.model import NautobotModel
+from nautobot_ssot.contrib.types import (
+    CustomFieldAnnotation,
+    CustomRelationshipAnnotation,
+)
+from nautobot_ssot.tests.contrib.typeddicts import DeviceDict, SoftwareImageFileDict, TagDict
+
+
+class LocationTypeModel(NautobotModel):
+    """Example model for LocationType in unittests."""
+
+    _modelname = "location_type"
+    _model = LocationType
+
+    name: str
+    nestable: Optional[bool]
+
+
+class DeviceModel(NautobotModel):
+    """Example model for unittests.
+
+    NOTE: We only need the typehints for this set of unittests.
+    """
+
+    _modelname = "device"
+    _model = Device
+
+    # Standard Attributes
+    name: str
+    vc_position: Optional[int]
+
+    # Foreign Keys
+    status__name: str
+    tenant__name: Optional[str]
+
+    # N to many Relationships
+    tags: List[TagDict] = []
+    software_image_files: Optional[List[SoftwareImageFileDict]]
+
+    # Custom Fields
+    custom_str: Annotated[str, CustomFieldAnnotation(name="custom_str")]
+    custom_int: Annotated[int, CustomFieldAnnotation(name="custom_int")]
+    custom_bool: Optional[Annotated[bool, CustomFieldAnnotation(name="custom_bool")]]
+
+    # Custom Foreign Keys
+    parent__name: Annotated[str, CustomRelationshipAnnotation(name="device_parent", side=RelationshipSideEnum.SOURCE)]
+
+    # Custom N to Many Relationships
+    children: Annotated[
+        List[DeviceDict],
+        CustomRelationshipAnnotation(name="device_children", side=RelationshipSideEnum.DESTINATION),
+    ]
+
+    # Invalid Fields
+    invalid_field: str