Initial commit

netbox_services/engine/allocators.py

@@ -0,0 +1,78 @@
+"""Allocators create or acquire objects that don't exist yet (e.g. IP addresses)."""
+
+from .registry import register_allocator, get_selector
+
+
+@register_allocator("first_free")
+class FirstFreeIPAddressAllocator:
+    """Allocate the first available IP address from a resolved prefix.
+
+    Spec shape::
+
+        {"strategy": "first_free", "prefix": {"type": "reference", "path": "parent.site.mgmt_prefix"}}
+        {"strategy": "first_free", "prefix": {"type": "prefix", "filters": {"site": "hq"}}}
+    """
+
+    def __init__(self, spec):
+        self.spec = spec
+        self.prefix_spec = spec.get("prefix", {})
+
+    def resolve_prefix(self, context):
+        from ipam.models import Prefix
+        selector = get_selector(self.prefix_spec)
+        qs = selector.resolve(context, Prefix)
+        return qs.first()
+
+    def allocate(self, context):
+        from ipam.models import IPAddress
+        from ipam.choices import IPAddressStatusChoices
+
+        prefix = self.resolve_prefix(context)
+        if prefix is None:
+            raise ValueError("No prefix resolved for first_free allocation.")
+
+        available = prefix.get_first_available_ip()
+        if available is None:
+            raise ValueError(
+                f"No available IP addresses in prefix {prefix.prefix}."
+            )
+
+        ip = IPAddress.objects.create(
+            address=available,
+            status=IPAddressStatusChoices.STATUS_ACTIVE,
+        )
+        return ip
+
+
+@register_allocator("specific")
+class SpecificAllocator:
+    """Use a specific object identified by content type + ID or context path.
+
+    Spec shape (by reference)::
+
+        {"strategy": "specific", "path": "parent.site.primary_ip4"}
+
+    Spec shape (by value, for IP addresses)::
+
+        {"strategy": "specific", "address": "10.0.0.1/24"}
+    """
+
+    def __init__(self, spec):
+        self.spec = spec
+        self.path = spec.get("path")
+        self.address = spec.get("address")
+
+    def allocate(self, context):
+        if self.address:
+            from ipam.models import IPAddress
+            from ipam.choices import IPAddressStatusChoices
+            ip, _ = IPAddress.objects.get_or_create(
+                address=self.address,
+                defaults={"status": IPAddressStatusChoices.STATUS_ACTIVE},
+            )
+            return ip
+
+        if self.path:
+            return context.resolve_path(self.path)
+
+        raise ValueError("SpecificAllocator requires 'path' or 'address'.")