7699 Add Scope to Cluster

docs/models/virtualization/cluster.md

@@ -23,6 +23,6 @@ The cluster's operational status.
 !!! tip
     Additional statuses may be defined by setting `Cluster.status` under the [`FIELD_CHOICES`](../../configuration/data-validation.md#field_choices) configuration parameter.
 
-### Site
+### Scope
 
-The [site](../dcim/site.md) with which the cluster is associated.
+The [region](../dcim/region.md), [site](../dcim/site.md), [site group](../dcim/sitegroup.md) or [location](../dcim/location.md) with which this cluster is associated.

netbox/dcim/constants.py

@@ -123,3 +123,8 @@
     'powerport': ['poweroutlet', 'powerfeed'],
     'rearport': ['consoleport', 'consoleserverport', 'interface', 'frontport', 'rearport', 'circuittermination'],
 }
+
+# Models which can serve to scope an object by location
+LOCATION_SCOPE_TYPES = (
+    'region', 'sitegroup', 'site', 'location',
+)

netbox/dcim/filtersets.py

@@ -73,6 +73,7 @@
     'RearPortFilterSet',
     'RearPortTemplateFilterSet',
     'RegionFilterSet',
+    'ScopedFilterSet',
     'SiteFilterSet',
     'SiteGroupFilterSet',
     'VirtualChassisFilterSet',
@@ -2344,3 +2345,60 @@ class InterfaceConnectionFilterSet(ConnectionFilterSet):
     class Meta:
         model = Interface
         fields = tuple()
+
+
+class ScopedFilterSet(BaseFilterSet):
+    """
+    Provides additional filtering functionality for location, site, etc.. for Scoped models.
+    """
+    scope_type = ContentTypeFilter()
+    region_id = TreeNodeMultipleChoiceFilter(
+        queryset=Region.objects.all(),
+        field_name='_region',
+        lookup_expr='in',
+        label=_('Region (ID)'),
+    )
+    region = TreeNodeMultipleChoiceFilter(
+        queryset=Region.objects.all(),
+        field_name='_region',
+        lookup_expr='in',
+        to_field_name='slug',
+        label=_('Region (slug)'),
+    )
+    site_group_id = TreeNodeMultipleChoiceFilter(
+        queryset=SiteGroup.objects.all(),
+        field_name='_site_group',
+        lookup_expr='in',
+        label=_('Site group (ID)'),
+    )
+    site_group = TreeNodeMultipleChoiceFilter(
+        queryset=SiteGroup.objects.all(),
+        field_name='_site_group',
+        lookup_expr='in',
+        to_field_name='slug',
+        label=_('Site group (slug)'),
+    )
+    site_id = django_filters.ModelMultipleChoiceFilter(
+        queryset=Site.objects.all(),
+        field_name='_site',
+        label=_('Site (ID)'),
+    )
+    site = django_filters.ModelMultipleChoiceFilter(
+        field_name='_site__slug',
+        queryset=Site.objects.all(),
+        to_field_name='slug',
+        label=_('Site (slug)'),
+    )
+    location_id = TreeNodeMultipleChoiceFilter(
+       queryset=Location.objects.all(),
+       field_name='_location',
+       lookup_expr='in',
+       label=_('Location (ID)'),
+    )
+    location = TreeNodeMultipleChoiceFilter(
+        queryset=Location.objects.all(),
+        field_name='_location',
+        lookup_expr='in',
+        to_field_name='slug',
+        label=_('Location (slug)'),
+    )

netbox/dcim/forms/mixins.py

@@ -0,0 +1,105 @@
+from django import forms
+from django.contrib.contenttypes.models import ContentType
+from django.core.exceptions import ObjectDoesNotExist
+from django.utils.translation import gettext_lazy as _
+
+from dcim.constants import LOCATION_SCOPE_TYPES
+from dcim.models import Site
+from utilities.forms import get_field_value
+from utilities.forms.fields import (
+    ContentTypeChoiceField, CSVContentTypeField, DynamicModelChoiceField,
+)
+from utilities.templatetags.builtins.filters import bettertitle
+from utilities.forms.widgets import HTMXSelect
+
+__all__ = (
+    'ScopedBulkEditForm',
+    'ScopedForm',
+    'ScopedImportForm',
+)
+
+
+class ScopedForm(forms.Form):
+    scope_type = ContentTypeChoiceField(
+        queryset=ContentType.objects.filter(model__in=LOCATION_SCOPE_TYPES),
+        widget=HTMXSelect(),
+        required=False,
+        label=_('Scope type')
+    )
+    scope = DynamicModelChoiceField(
+        label=_('Scope'),
+        queryset=Site.objects.none(),  # Initial queryset
+        required=False,
+        disabled=True,
+        selector=True
+    )
+
+    def __init__(self, *args, **kwargs):
+        instance = kwargs.get('instance')
+        initial = kwargs.get('initial', {})
+
+        if instance is not None and instance.scope:
+            initial['scope'] = instance.scope
+            kwargs['initial'] = initial
+
+        super().__init__(*args, **kwargs)
+        self._set_scoped_values()
+
+    def clean(self):
+        super().clean()
+
+        # Assign the selected scope (if any)
+        self.instance.scope = self.cleaned_data.get('scope')
+
+    def _set_scoped_values(self):
+        if scope_type_id := get_field_value(self, 'scope_type'):
+            try:
+                scope_type = ContentType.objects.get(pk=scope_type_id)
+                model = scope_type.model_class()
+                self.fields['scope'].queryset = model.objects.all()
+                self.fields['scope'].widget.attrs['selector'] = model._meta.label_lower
+                self.fields['scope'].disabled = False
+                self.fields['scope'].label = _(bettertitle(model._meta.verbose_name))
+            except ObjectDoesNotExist:
+                pass
+
+            if self.instance and scope_type_id != self.instance.scope_type_id:
+                self.initial['scope'] = None
+
+
+class ScopedBulkEditForm(forms.Form):
+    scope_type = ContentTypeChoiceField(
+        queryset=ContentType.objects.filter(model__in=LOCATION_SCOPE_TYPES),
+        widget=HTMXSelect(method='post', attrs={'hx-select': '#form_fields'}),
+        required=False,
+        label=_('Scope type')
+    )
+    scope = DynamicModelChoiceField(
+        label=_('Scope'),
+        queryset=Site.objects.none(),  # Initial queryset
+        required=False,
+        disabled=True,
+        selector=True
+    )
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        if scope_type_id := get_field_value(self, 'scope_type'):
+            try:
+                scope_type = ContentType.objects.get(pk=scope_type_id)
+                model = scope_type.model_class()
+                self.fields['scope'].queryset = model.objects.all()
+                self.fields['scope'].widget.attrs['selector'] = model._meta.label_lower
+                self.fields['scope'].disabled = False
+                self.fields['scope'].label = _(bettertitle(model._meta.verbose_name))
+            except ObjectDoesNotExist:
+                pass
+
+
+class ScopedImportForm(forms.Form):
+    scope_type = CSVContentTypeField(
+        queryset=ContentType.objects.filter(model__in=LOCATION_SCOPE_TYPES),
+        required=False,
+        label=_('Scope type (app & model)')
+    )

netbox/dcim/graphql/types.py

@@ -463,6 +463,10 @@ class LocationType(VLANGroupsMixin, ImageAttachmentsMixin, ContactsMixin, Organi
     devices: List[Annotated["DeviceType", strawberry.lazy('dcim.graphql.types')]]
     children: List[Annotated["LocationType", strawberry.lazy('dcim.graphql.types')]]
 
+    @strawberry_django.field
+    def clusters(self) -> List[Annotated["ClusterType", strawberry.lazy('virtualization.graphql.types')]]:
+        return self._clusters.all()
+
     @strawberry_django.field
     def circuit_terminations(self) -> List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]:
         return self.circuit_terminations.all()
@@ -710,6 +714,10 @@ class RegionType(VLANGroupsMixin, ContactsMixin, OrganizationalObjectType):
     def parent(self) -> Annotated["RegionType", strawberry.lazy('dcim.graphql.types')] | None:
         return self.parent
 
+    @strawberry_django.field
+    def clusters(self) -> List[Annotated["ClusterType", strawberry.lazy('virtualization.graphql.types')]]:
+        return self._clusters.all()
+
     @strawberry_django.field
     def circuit_terminations(self) -> List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]:
         return self.circuit_terminations.all()
@@ -735,9 +743,14 @@ class SiteType(VLANGroupsMixin, ImageAttachmentsMixin, ContactsMixin, NetBoxObje
     devices: List[Annotated["DeviceType", strawberry.lazy('dcim.graphql.types')]]
     locations: List[Annotated["LocationType", strawberry.lazy('dcim.graphql.types')]]
     asns: List[Annotated["ASNType", strawberry.lazy('ipam.graphql.types')]]
+    circuit_terminations: List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]
     clusters: List[Annotated["ClusterType", strawberry.lazy('virtualization.graphql.types')]]
     vlans: List[Annotated["VLANType", strawberry.lazy('ipam.graphql.types')]]
 
+    @strawberry_django.field
+    def clusters(self) -> List[Annotated["ClusterType", strawberry.lazy('virtualization.graphql.types')]]:
+        return self._clusters.all()
+
     @strawberry_django.field
     def circuit_terminations(self) -> List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]:
         return self.circuit_terminations.all()
@@ -758,6 +771,10 @@ class SiteGroupType(VLANGroupsMixin, ContactsMixin, OrganizationalObjectType):
     def parent(self) -> Annotated["SiteGroupType", strawberry.lazy('dcim.graphql.types')] | None:
         return self.parent
 
+    @strawberry_django.field
+    def clusters(self) -> List[Annotated["ClusterType", strawberry.lazy('virtualization.graphql.types')]]:
+        return self._clusters.all()
+
     @strawberry_django.field
     def circuit_terminations(self) -> List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]:
         return self.circuit_terminations.all()

netbox/dcim/models/devices.py

@@ -958,10 +958,17 @@ def clean(self):
                 })
 
         # A Device can only be assigned to a Cluster in the same Site (or no Site)
-        if self.cluster and self.cluster.site is not None and self.cluster.site != self.site:
+        if self.cluster and self.cluster._site is not None and self.cluster._site != self.site:
             raise ValidationError({
                 'cluster': _("The assigned cluster belongs to a different site ({site})").format(
-                    site=self.cluster.site
+                    site=self.cluster._site
+                )
+            })
+
+        if self.cluster and self.cluster._location is not None and self.cluster._location != self.location:
+            raise ValidationError({
+                'cluster': _("The assigned cluster belongs to a different location ({location})").format(
+                    site=self.cluster._location
                 )
             })
 

netbox/dcim/models/mixins.py

@@ -1,6 +1,10 @@
+from django.apps import apps
+from django.contrib.contenttypes.fields import GenericForeignKey
 from django.db import models
+from dcim.constants import LOCATION_SCOPE_TYPES
 
 __all__ = (
+    'CachedScopeMixin',
     'RenderConfigMixin',
 )
 
@@ -27,3 +31,84 @@ def get_config_template(self):
             return self.role.config_template
         if self.platform and self.platform.config_template:
             return self.platform.config_template
+
+
+class CachedScopeMixin(models.Model):
+    """
+    Mixin for adding a GenericForeignKey scope to a model that can point to a Region, SiteGroup, Site, or Location.
+    Includes cached fields for each to allow efficient filtering. Appropriate validation must be done in the clean()
+    method as this does not have any as validation is generally model-specific.
+    """
+    scope_type = models.ForeignKey(
+        to='contenttypes.ContentType',
+        on_delete=models.PROTECT,
+        limit_choices_to=models.Q(model__in=LOCATION_SCOPE_TYPES),
+        related_name='+',
+        blank=True,
+        null=True
+    )
+    scope_id = models.PositiveBigIntegerField(
+        blank=True,
+        null=True
+    )
+    scope = GenericForeignKey(
+        ct_field='scope_type',
+        fk_field='scope_id'
+    )
+
+    _location = models.ForeignKey(
+        to='dcim.Location',
+        on_delete=models.CASCADE,
+        related_name='_%(class)ss',
+        blank=True,
+        null=True
+    )
+    _site = models.ForeignKey(
+        to='dcim.Site',
+        on_delete=models.CASCADE,
+        related_name='_%(class)ss',
+        blank=True,
+        null=True
+    )
+    _region = models.ForeignKey(
+        to='dcim.Region',
+        on_delete=models.CASCADE,
+        related_name='_%(class)ss',
+        blank=True,
+        null=True
+    )
+    _site_group = models.ForeignKey(
+        to='dcim.SiteGroup',
+        on_delete=models.CASCADE,
+        related_name='_%(class)ss',
+        blank=True,
+        null=True
+    )
+
+    class Meta:
+        abstract = True
+
+    def save(self, *args, **kwargs):
+        # Cache objects associated with the terminating object (for filtering)
+        self.cache_related_objects()
+
+        super().save(*args, **kwargs)
+
+    def cache_related_objects(self):
+        self._region = self._site_group = self._site = self._location = None
+        if self.scope_type:
+            scope_type = self.scope_type.model_class()
+            if scope_type == apps.get_model('dcim', 'region'):
+                self._region = self.scope
+            elif scope_type == apps.get_model('dcim', 'sitegroup'):
+                self._site_group = self.scope
+            elif scope_type == apps.get_model('dcim', 'site'):
+                self._region = self.scope.region
+                self._site_group = self.scope.group
+                self._site = self.scope
+            elif scope_type == apps.get_model('dcim', 'location'):
+                self._region = self.scope.site.region
+                self._site_group = self.scope.site.group
+                self._site = self.scope.site
+                self._location = self.scope
+    cache_related_objects.alters_data = True

netbox/dcim/tests/test_models.py

@@ -601,11 +601,12 @@ def test_device_mismatched_site_cluster(self):
         Site.objects.bulk_create(sites)
 
         clusters = (
-            Cluster(name='Cluster 1', type=cluster_type, site=sites[0]),
-            Cluster(name='Cluster 2', type=cluster_type, site=sites[1]),
-            Cluster(name='Cluster 3', type=cluster_type, site=None),
+            Cluster(name='Cluster 1', type=cluster_type, scope=sites[0]),
+            Cluster(name='Cluster 2', type=cluster_type, scope=sites[1]),
+            Cluster(name='Cluster 3', type=cluster_type, scope=None),
         )
-        Cluster.objects.bulk_create(clusters)
+        for cluster in clusters:
+            cluster.save()
 
         device_type = DeviceType.objects.first()
         device_role = DeviceRole.objects.first()

netbox/extras/tests/test_models.py

@@ -274,7 +274,7 @@ def test_annotation_same_as_get_for_object_virtualmachine_relations(self):
             name="Cluster",
             group=cluster_group,
             type=cluster_type,
-            site=site,
+            scope=site,
         )
 
         region_context = ConfigContext.objects.create(
@@ -366,7 +366,7 @@ def test_virtualmachine_site_context(self):
         """
         site = Site.objects.first()
         cluster_type = ClusterType.objects.create(name="Cluster Type")
-        cluster = Cluster.objects.create(name="Cluster", type=cluster_type, site=site)
+        cluster = Cluster.objects.create(name="Cluster", type=cluster_type, scope=site)
         vm_role = DeviceRole.objects.first()
 
         # Create a ConfigContext associated with the site