105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
"""Selectors resolve a specification into a queryset of matching NetBox objects."""
|
|
|
|
import importlib
|
|
|
|
from django.db.models import Q
|
|
|
|
from .registry import register_selector
|
|
|
|
|
|
def _get_filterset_for_model(model):
|
|
"""Look up a model's FilterSet by convention: {app_label}.filtersets.{ModelName}FilterSet."""
|
|
app_label = model._meta.app_label
|
|
model_name = model._meta.model_name
|
|
class_name = f"{model.__name__}FilterSet"
|
|
try:
|
|
module = importlib.import_module(f"{app_label}.filtersets")
|
|
return getattr(module, class_name, None)
|
|
except (ImportError, AttributeError):
|
|
return None
|
|
|
|
|
|
@register_selector("netbox_filter")
|
|
class NetBoxFilterSelector:
|
|
"""Select objects by applying NetBox filter parameters.
|
|
|
|
Spec shape::
|
|
{"type": "netbox_filter", "filters": {"role": "router", "site__in": ["hq"]}}
|
|
"""
|
|
|
|
def __init__(self, spec):
|
|
self.spec = spec
|
|
self.filters = spec.get("filters", {})
|
|
|
|
def resolve(self, context, model):
|
|
qs = model.objects.all()
|
|
filterset_cls = _get_filterset_for_model(model)
|
|
if filterset_cls is not None:
|
|
fs = filterset_cls(self.filters, queryset=qs, request=None)
|
|
qs = fs.qs
|
|
else:
|
|
qs = qs.filter(**self._build_q(self.filters))
|
|
return qs
|
|
|
|
@staticmethod
|
|
def _build_q(filters):
|
|
"""Best-effort: turn the filter dict into ORM kwargs (no fancy lookups)."""
|
|
return {k: v for k, v in filters.items()}
|
|
|
|
|
|
@register_selector("reference")
|
|
class ReferenceSelector:
|
|
"""Select a single object via a dotted context path.
|
|
|
|
Spec shape::
|
|
{"type": "reference", "path": "parent.site"}
|
|
"""
|
|
|
|
def __init__(self, spec):
|
|
self.spec = spec
|
|
self.path = spec.get("path")
|
|
|
|
def resolve(self, context, model):
|
|
obj = context.resolve_path(self.path)
|
|
if obj is None:
|
|
return model.objects.none()
|
|
if isinstance(obj, (list, tuple)):
|
|
return model.objects.filter(pk__in=[o.pk for o in obj])
|
|
return model.objects.filter(pk=obj.pk)
|
|
|
|
|
|
@register_selector("prefix")
|
|
class PrefixSelector:
|
|
"""Select ipam.Prefix objects by filter or context reference.
|
|
|
|
Spec shape (filter)::
|
|
{"type": "prefix", "filters": {"site": "hq", "tag": "mgmt"}}
|
|
|
|
Spec shape (reference)::
|
|
{"type": "prefix", "path": "parent.site.management_prefix"}
|
|
"""
|
|
|
|
def __init__(self, spec):
|
|
self.spec = spec
|
|
self.filters = spec.get("filters")
|
|
self.path = spec.get("path")
|
|
|
|
def resolve(self, context, model):
|
|
from ipam.models import Prefix
|
|
|
|
if self.path:
|
|
obj = context.resolve_path(self.path)
|
|
if obj is None:
|
|
return Prefix.objects.none()
|
|
return Prefix.objects.filter(pk=obj.pk)
|
|
|
|
if self.filters is not None:
|
|
filterset_cls = _get_filterset_for_model(Prefix)
|
|
qs = Prefix.objects.all()
|
|
if filterset_cls is not None:
|
|
fs = filterset_cls(self.filters, queryset=qs, request=None)
|
|
return fs.qs
|
|
return qs.filter(**self.filters)
|
|
|
|
return Prefix.objects.none()
|