Change netbox payload typing so its aligned with singledispatch

Author: tjeerddie

products/product_blocks/nsistp.py

@@ -40,7 +40,7 @@ class NsistpBlockProvisioning(NsistpBlockInactive, lifecycle=[SubscriptionLifecy
     expose_in_topology: bool | None = None
     bandwidth: int | None = None
 
-    @computed_field
+    @computed_field  # type: ignore[prop-decorator]
     @property
     def title(self) -> str:
         return f"NSISTP {self.stp_id} on {self.sap.title}"

products/services/netbox/netbox.py

@@ -77,7 +77,7 @@ def _(model: CoreLinkBlockProvisioning, subscription: SubscriptionModel, **kwarg
 
 
 @build_payload.register
-def _(model: SAPBlockProvisioning, subscription: SubscriptionModel, **kwargs: Any) -> netbox.VlanPayload:
+def _(model: SAPBlockProvisioning, subscription: SubscriptionModel, **kwargs: Any) -> netbox.VlansPayload:
     return build_sap_payload(model, subscription)
 
 

products/services/netbox/payload/sap.py

@@ -16,11 +16,10 @@
 
 from products.product_blocks.sap import SAPBlockProvisioning
 from services import netbox
-import itertools
 
 
-def build_sap_payload(model: SAPBlockProvisioning, subscription: SubscriptionModel) -> list[netbox.VlanPayload]:
-    """Create and return a Netbox payload object for a :class:`~products.product_blocks.sap.SAPBlockProvisioning`.
+def build_sap_vlans_payload(model: SAPBlockProvisioning, subscription: SubscriptionModel) -> list[netbox.VlanPayload]:
+    """Create and return a list of Netbox :class:`VlanPayload` for a :class:`~products.product_blocks.sap.SAPBlockProvisioning`.
 
     Example payload::
 
@@ -35,16 +34,17 @@ def build_sap_payload(model: SAPBlockProvisioning, subscription: SubscriptionMod
         model: SAPBlockProvisioning
         subscription: The Subscription that will be provisioned
 
-    Returns: :class:`netbox.VlanPayload`
+    Returns: list[:class:`netbox.VlanPayload`]
 
     """
+    assert model.ims_id
     name = f"{model.port.node.node_name} {model.port.port_name}"
     vlan_list = [vlan for vlan_start, vlan_end in model.vlan.to_list_of_tuples() for vlan in range(vlan_start, vlan_end+1)]
     return [netbox.VlanPayload(vid=vlan, group=model.ims_id, name=f"{name} - {vlan}") for vlan in vlan_list]
 
 
 def build_sap_vlan_group_payload(model: SAPBlockProvisioning, subscription: SubscriptionModel) -> netbox.VlanGroupPayload:
-    """Create
+    """Create and return a list of Netbox :class:`VlanGroupPayload` for a :class:`~products.product_blocks.sap.SAPBlockProvisioning`.
 
     Example payload::
 
@@ -63,3 +63,26 @@ def build_sap_vlan_group_payload(model: SAPBlockProvisioning, subscription: Subs
     name = f"{model.port.node.node_name} {model.port.port_name}"
     slug = name.replace(" ", "-").replace("/", "-")
     return netbox.VlanGroupPayload(name=name, slug=slug, vid_ranges=model.vlan.to_list_of_tuples())
+
+
+def build_sap_payload(model: SAPBlockProvisioning, subscription: SubscriptionModel) -> netbox.VlansPayload:
+    """Create and return a Netbox payload object for a :class:`~products.product_blocks.sap.SAPBlockProvisioning`.
+
+    Example payload::
+
+       {
+          "vid": 4,
+          "name": "paris01a 0/0/1 vlan 4",
+          "status": "active",
+          "group": 1
+       }
+
+    Args:
+        model: SAPBlockProvisioning
+        subscription: The Subscription that will be provisioned
+
+    Returns: :class:`netbox.VlanPayload`
+
+    """
+    vlan_payloads = build_sap_vlans_payload(model, subscription)
+    return netbox.VlansPayload(vlans=vlan_payloads)

services/netbox.py

@@ -130,7 +130,7 @@ class AvailableIpPayload:
 
 @dataclass
 class VlanPayload(NetboxPayload):
-    vid: OrchestratorVlanRanges
+    vid: int
     name: str
     group: int
     status: str | None = "active"
@@ -140,7 +140,14 @@ class VlanPayload(NetboxPayload):
 class VlanGroupPayload(NetboxPayload):
     name: str
     slug: str
-    vid_ranges: list[list[int, int]]
+    vid_ranges: list[tuple[int, int]]
+
+
+@dataclass
+class VlansPayload(NetboxPayload):
+    vlans: list[VlanPayload]
+
+
 
 
 @dataclass
@@ -428,6 +435,13 @@ def _(payload: VlanPayload, **kwargs: Any) -> int:
     return _create_object(payload, endpoint=api.ipam.vlans)
 
 
+@create.register
+def _(payload: VlansPayload, **kwargs: Any) -> int:
+    for vlan_payload in payload.vlans:
+        _create_object(vlan_payload, endpoint=api.ipam.vlans)
+    return payload.vlans[0].group
+
+
 @create.register
 def _(payload: VlanGroupPayload, **kwargs: Any) -> int:
     return _create_object(payload, endpoint=api.ipam.vlan_groups)

workflows/l2vpn/create_l2vpn.py

@@ -109,8 +109,7 @@ def ims_create_vlans(subscription: L2vpnProvisioning) -> State:
     for sap in subscription.virtual_circuit.saps:
         group_payload = build_sap_vlan_group_payload(sap, subscription)
         sap.ims_id = netbox.create(group_payload)
-        group_payloads = group_payloads + [group_payload]
-
+        group_payloads += [group_payload]
         vlan_payloads += build_payload(sap, subscription)
 
     for payload in vlan_payloads:

workflows/nsistp/create_nsistp.py

@@ -30,7 +30,6 @@
 
 from products.product_types.nsistp import NsistpInactive, NsistpProvisioning
 from products.services.description import description
-from products.services.netbox.netbox import build_payload
 from products.services.netbox.payload.sap import build_sap_vlan_group_payload
 from services import netbox
 from workflows.nsistp.shared.forms import (

workflows/nsistp/modify_nsistp.py

@@ -35,7 +35,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     nsistp = subscription.nsistp
 
     class ModifyNsistpForm(FormPage):
-        stp_id: read_only_field(nsistp.stp_id)
+        stp_id: read_only_field(nsistp.stp_id)  # type: ignore[valid-type]
 
         divider_1: Divider
 

workflows/nsistp/shared/forms.py

@@ -57,7 +57,7 @@ def port_selector() -> type[Choice]:
         str(subscription.subscription_id): subscription.description
         for subscription in sorted(port_subscriptions, key=lambda port: port.description)
     }
-    return Choice("Port", zip(ports.keys(), ports.items()))
+    return Choice("Port", zip(ports.keys(), ports.items()))  # type: ignore
 
 
 def is_fqdn(hostname: str) -> bool:
@@ -124,11 +124,10 @@ def validate_stp_id_uniqueness(subscription_id: UUID | None, stp_id: str, info:
     topology = values.get("topology")
 
     if customer_id and topology:
-
         def is_not_unique(nsistp: Nsistp) -> bool:
             return (
-                nsistp.settings.stp_id.casefold() == stp_id.casefold()
-                and nsistp.settings.topology.casefold() == topology.casefold()
+                nsistp.nsistp.stp_id.casefold() == stp_id.casefold()
+                and nsistp.nsistp.topology.casefold() == topology.casefold()
             )
 
         subscriptions = _get_nsistp_subscriptions(subscription_id)
@@ -152,7 +151,7 @@ def validate_nurn(nurn: str | None) -> str | None:
     return nurn
 
 
-def nsistp_fill_sap(subscription: NsistpInactive, subscription_id: UUIDstr, vlan: OrchestratorVlanRanges | int) -> None:
+def nsistp_fill_sap(subscription: NsistpInactive, subscription_id: UUIDstr, vlan: OrchestratorVlanRanges) -> None:
     subscription.nsistp.sap.vlan = vlan
     subscription.nsistp.sap.port = SubscriptionModel.from_subscription(subscription_id).port  # type: ignore
 

workflows/nsistp/shared/vlan.py

@@ -64,13 +64,11 @@ def validate_vlan_not_in_use(
 
     Returns: input value if no errors
     """
-    print(port_field_name, info.data.get(port_field_name))
     if not (subscription_id := info.data.get(port_field_name)):
         return vlan
 
     used_vlans = find_allocated_vlans(subscription_id)
 
-    print(used_vlans)
     # Remove currently chosen vlans for this port to prevent tripping on in used by itself
     current_selected_vlan_ranges: list[str] = []
     if current:
@@ -82,7 +80,7 @@ def validate_vlan_not_in_use(
                 current_selected_vlan = "0"
 
             current_selected_vlan_range = OrchestratorVlanRanges(current_selected_vlan)
-            used_vlans -= current_selected_vlan_range
+            used_vlans -= current_selected_vlan_range  # type: ignore[assignment]
             current_selected_vlan_ranges = [
                 *current_selected_vlan_ranges,
                 *list(current_selected_vlan_range),
@@ -103,7 +101,7 @@ def validate_vlan_not_in_use(
 
 def find_allocated_vlans(
     subscription_id: UUID | UUIDstr,
-) -> list[int]:
+) -> OrchestratorVlanRanges:
     """Find all vlans already allocated to a SAP for a given port."""
     logger.debug(
         "Finding allocated VLANs",

workflows/shared.py

@@ -132,7 +132,7 @@ def modify_summary_form(user_input: dict, block: ProductBlockModel, fields: List
     before = [str(getattr(block, nm)) for nm in fields]  # type: ignore[attr-defined]
     after = [str(user_input[nm]) for nm in fields]
     yield from summary_form(
-        block.subscription.product.name,
+        block.subscription.product.name if block.subscription else "No Product Name Found",
         SummaryData(labels=fields, headers=["Before", "After"], columns=[before, after]),
     )