import json import yaml from django import forms from django.core.exceptions import ValidationError from django.utils.translation import gettext_lazy as _ from netbox.forms import ( NetBoxModelBulkEditForm, NetBoxModelFilterSetForm, NetBoxModelForm, NetBoxModelImportForm, PrimaryModelForm, ) from tenancy.models import Tenant from utilities.forms.fields import ( DynamicModelChoiceField, DynamicModelMultipleChoiceField, TagFilterField, ) from utilities.forms.rendering import FieldSet from .models import ( Service, ServiceStatusChoices, ServiceTemplate, ServiceTemplateStatusChoices, validate_tree, ) # ── Service Template ───────────────────────────────────────────────── EXAMPLE_TREE_YAML = """\ # Requirement tree — defines what objects a service needs. # Delete this example and write your own. YAML or JSON accepted. # # Node parameters: # name (required) unique identifier within the tree # object_type (required) NetBox model as "app_label.model_name" # quantity (optional, default 1) how many objects to resolve # scope (optional, default "parent") # "parent" = each parent gets its own children; # "service" = children resolved once at service # level (only meaningful when quantity > 1) # selector (mutually exclusive with allocation) find # existing objects # allocation (mutually exclusive with selector) create/acquire # objects # children (optional) list of child requirement nodes # # Selector types: # netbox_filter — filter existing objects via NetBox filter # params # reference — resolve a single object via context path # (parent.site) # prefix — select ipam.Prefix by filter or reference # # Allocation strategies: # create — instantiate a new model from attrs # (returned unsaved, resolver saves after # auto-linking to parent) # first_free — allocate next available IP from a resolved # prefix # specific — use an existing object by address or context # path # # Jinja2 templating in allocation.attrs values: # {{ service.tenant.slug }} — the service's tenant # {{ index + 1 }} — 0-based index within a quantity # loop # {{ ..name }} — immediate parent's name # (.. = parent) # {{ ...site.slug }} — grandparent's site slug # (... = grandparent) # {{ ....region.slug }} — great-grandparent (.... = p3) # FK fields must be PKs (integers), e.g. site: 1 or # site: "{{ ..site.pk }}" # # Auto-linking: children created via create/first_free # are automatically linked to their parent via the central mapper # (engine/mappers.py). # You do NOT specify the link field — the mapper knows. # If a mapping is missing, resolution fails with an error telling # you exactly which register_link() call to add. # ── Example 1: Select 2 existing routers, assign each a management IP ── - name: routers object_type: dcim.device quantity: 2 selector: type: netbox_filter filters: role: router site: hq scope: service children: - name: mgmt_ip object_type: ipam.ipaddress quantity: 1 allocation: strategy: first_free prefix: type: prefix filters: prefix: 10.0.0.0/8 scope: parent # ── Example 2: Create a new device + interface + IP (all auto-linked) ── - name: edge_router object_type: dcim.device quantity: 1 allocation: strategy: create attrs: name: "rtr-{{ service.tenant.slug }}-{{ index + 1 }}" device_type: 1 # PK of a DeviceType role: 1 # PK of a DeviceRole site: "{{ ..site.pk }}" # not valid at top-level (no parent); # use a literal PK here, e.g. site: 1 status: active scope: service children: - name: uplink object_type: dcim.interface quantity: 1 allocation: strategy: create attrs: name: eth0 type: 1000base-t scope: parent children: - name: uplink_ip object_type: ipam.ipaddress quantity: 1 allocation: strategy: first_free prefix: type: prefix filters: prefix: 10.0.0.0/8 scope: parent # ── Example 3: Use a specific existing IP address ── - name: loopback object_type: ipam.ipaddress quantity: 1 allocation: strategy: specific address: 10.255.255.1/32 scope: service """ class ServiceTemplateForm(PrimaryModelForm): tree_raw = forms.CharField( label=_("Requirement tree"), required=False, widget=forms.Textarea( attrs={ "rows": 20, "class": "font-monospace", "placeholder": ( '- name: routers\n' ' object_type: dcim.device\n' ' quantity: 2\n' ' selector:\n' ' type: netbox_filter\n' ' filters:\n' ' role: router\n' ' children:\n' ' - name: mgmt-ip\n' ' object_type: ipam.ipaddress\n' ' allocation:\n' ' strategy: first_free\n' ' prefix:\n' ' type: prefix\n' ' filters: {}\n' ), }, ), help_text=_("Enter the requirement tree as YAML or JSON."), ) fieldsets = ( FieldSet("name", "status", "tree_raw", name=_("Service Template")), FieldSet("description", "tags", name=_("Metadata")), ) class Meta: model = ServiceTemplate fields = ("name", "status", "description", "comments") def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) if self.instance and self.instance.pk and self.instance.tree: self.fields["tree_raw"].initial = yaml.dump( self.instance.tree, default_flow_style=False, sort_keys=False, ) elif not self.instance.pk: self.fields["tree_raw"].initial = EXAMPLE_TREE_YAML def clean(self): super().clean() raw = self.cleaned_data.get("tree_raw", "").strip() if not raw: self.instance.tree = [] return self.cleaned_data tree = None try: tree = json.loads(raw) except (json.JSONDecodeError, ValueError): pass if tree is None: try: tree = yaml.safe_load(raw) except yaml.YAMLError as exc: self.add_error("tree_raw", _(f"Invalid YAML or JSON: {exc}")) return self.cleaned_data if tree is None: tree = [] self.instance.tree = tree self.instance._tree_validated_by_form = True try: validate_tree(tree) except ValidationError as exc: self.add_error("tree_raw", ValidationError(exc.messages)) return self.cleaned_data class ServiceTemplateFilterSetForm(NetBoxModelFilterSetForm): model = ServiceTemplate name = forms.CharField(required=False) status = forms.MultipleChoiceField( choices=ServiceTemplateStatusChoices, required=False, ) tag = TagFilterField(model) class ServiceTemplateBulkEditForm(NetBoxModelBulkEditForm): model = ServiceTemplate status = forms.ChoiceField( choices=ServiceTemplateStatusChoices, required=False, ) description = forms.CharField(required=False) nullable_fields = ("description",) class ServiceTemplateImportForm(NetBoxModelImportForm): class Meta: model = ServiceTemplate fields = ("name", "status", "description", "comments", "tags") # ── Service ────────────────────────────────────────────────────────── class ServiceForm(PrimaryModelForm): service_template = DynamicModelChoiceField( queryset=ServiceTemplate.objects.all(), ) tenant = DynamicModelChoiceField( queryset=Tenant.objects.all(), ) fieldsets = ( FieldSet("name", "service_template", "tenant", "status", "description", "tags", name=_("Service")), ) class Meta: model = Service fields = ("name", "service_template", "tenant", "status", "description", "comments") class ServiceFilterSetForm(NetBoxModelFilterSetForm): model = Service name = forms.CharField(required=False) service_template = DynamicModelMultipleChoiceField( queryset=ServiceTemplate.objects.all(), required=False, ) tenant = DynamicModelMultipleChoiceField( queryset=Tenant.objects.all(), required=False, ) status = forms.MultipleChoiceField( choices=ServiceStatusChoices, required=False, ) tag = TagFilterField(model) class ServiceBulkEditForm(NetBoxModelBulkEditForm): model = Service status = forms.ChoiceField( choices=ServiceStatusChoices, required=False, ) description = forms.CharField(required=False) nullable_fields = ("description",) # ── Resolve form ───────────────────────────────────────────────────── class ResolveForm(forms.Form): mode = forms.ChoiceField( choices=( ("dry_run", _("Dry Run")), ("commit", _("Commit")), ), initial="dry_run", widget=forms.RadioSelect, )