rename some variables

dev/zones/example.com.yaml

@@ -1,11 +1,10 @@
 ---
 ? ''
-  :
+: ttl: 172800
-  ttl: 172800
   type: NS
   values:
-    - ns1.example.com.
+  - ns1.example.com.
-    - ns2.example.com.
+  - ns2.example.com.
 ns1:
   type: A
   value: 192.168.1.1

dev/zones/test.example.com.yaml

@@ -1,11 +1,10 @@
 ---
 ? ''
-  :
+: ttl: 172800
-  ttl: 172800
   type: NS
   values:
-    - ns1.example.com.
+  - ns1.example.com.
-    - ns2.example.com.
+  - ns2.example.com.
 record1:
   type: CNAME
   value: record11.test.example.com.

dev/zones/view.example.com.yaml

@@ -1,11 +1,10 @@
 ---
 ? ''
-  :
+: ttl: 172800
-  ttl: 172800
   type: NS
   values:
-    - ns1.example.com.
+  - ns1.example.com.
-    - ns2.example.com.
+  - ns2.example.com.
 record1:
   type: CNAME
   value: record11.view.example.com.
@@ -14,4 +13,4 @@ record11:
   value: 192.168.1.11
 record12:
   type: A
-  value: 192.168.1.12
+  value: 192.168.1.13

src/octodns_netbox_dns/__init__.py

@@ -11,9 +11,7 @@ import pynetbox.core.response
 
 
 class NetBoxDNSSource(octodns.provider.base.BaseProvider):
-    """
+    """OctoDNS provider for NetboxDNS"""
-    OctoDNS provider for NetboxDNS
-    """
 
     SUPPORTS_GEO = False
     SUPPORTS_DYNAMIC = False
@@ -62,9 +60,7 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
         *args,
         **kwargs,
     ):
-        """
+        """initialize the NetboxDNSSource"""
-        Initialize the NetboxDNSSource
-        """
         self.log = logging.getLogger(f"NetboxDNSSource[{id}]")
         self.log.debug(f"__init__: {id=}, {url=}, {view=}, {replace_duplicates=}, {make_absolute=}")
         super().__init__(id, *args, **kwargs)
@@ -76,8 +72,7 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
         self.make_absolute = make_absolute
 
     def _make_absolute(self, value: str) -> str:
-        """
+        """return dns name with trailing dot to make it absolute
-        Return dns name with trailing dot to make it absolute
 
         @param value: dns record value
 
@@ -92,8 +87,7 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
         return absolute_value
 
     def _get_nb_view(self, view: str | None | Literal[False]) -> dict[str, int | str]:
-        """
+        """get the correct netbox view when requested
-        Get the correct netbox view when requested
 
         @param view: `False` for no view, `None` for zones without a view, else the view name
 
@@ -115,8 +109,7 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
         return {"view_id": nb_view.id}
 
     def _get_nb_zone(self, name: str, view: dict[str, str | int]) -> pynetbox.core.response.Record:
-        """
+        """given a zone name and a view name, look it up in NetBox.
-        Given a zone name and a view name, look it up in NetBox.
 
         @param name: name of the dns zone
         @param view: the netbox view id in the api query format
@@ -132,15 +125,17 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
 
         return nb_zone
 
-    def _format_rdata(self, rdata: dns.rdata.Rdata, raw_value: str) -> str | dict[str, Any]:
+    def _format_rdata(
-        """
+        self, nb_record: pynetbox.core.response.Record, raw_value: str
-        Format netbox record values to correct octodns record values
+    ) -> str | dict[str, Any]:
+        """format netbox record values to correct octodns record values
 
-        @param rdata: rrdata record value
+        @param nb_record: netbox record
         @param raw_value: raw record value
 
         @return: formatted rrdata value
         """
+        rdata = dns.rdata.from_text("IN", nb_record.type, raw_value)
         match rdata.rdtype.name:
             case "A" | "AAAA":
                 value = rdata.address
@@ -221,8 +216,7 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
         return value  # type:ignore
 
     def _format_nb_records(self, zone: octodns.zone.Zone) -> list[dict[str, Any]]:
-        """
+        """format netbox dns records to the octodns format
-        Format netbox dns records to the octodns format
 
         @param zone: octodns zone
 
@@ -255,9 +249,8 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
 
             self.log.debug(f"record data={rcd_data}")
 
-            rdata = dns.rdata.from_text("IN", nb_record.type, raw_value)
             try:
-                rcd_value = self._format_rdata(rdata, raw_value)
+                rcd_value = self._format_rdata(nb_record, raw_value)
             except NotImplementedError:
                 continue
             except Exception as exc:
@@ -273,12 +266,13 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
     def populate(
         self, zone: octodns.zone.Zone, target: bool = False, lenient: bool = False
     ) -> bool:
-        """
+        """get all the records of a zone from NetBox and add them to the OctoDNS zone
-        Get all the records of a zone from NetBox and add them to the OctoDNS zone
 
         @param zone: octodns zone
         @param target: when `True`, load the current state of the provider.
         @param lenient: when `True`, skip record validation and do a "best effort" load of data.
+
+        @return: true if the zone exists, else false.
         """
         self.log.info(f"populate -> '{zone.name}', target={target}, lenient={lenient}")
 
@@ -306,118 +300,114 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
 
         return True
 
-    def _include_change(self, change: octodns.record.change.Change) -> bool:
+    @staticmethod
-        """Filter out record types which the provider can't create in netbox"""
+    def _format_changeset(change: Any) -> set[str]:
-        if change.new._type in ["SOA", "PTR", "NS"]:
+        """format the changeset
+
+        @param change: the raw changes
+
+        @return: the formatted changeset
+        """
+        match change:
+            case octodns.record.ValueMixin():
+                changeset = {repr(change.value)[1:-1]}
+            case octodns.record.ValuesMixin():
+                changeset = {repr(v)[1:-1] for v in change.values}
+            case _:
+                raise ValueError
+
+        return changeset
+
+    @staticmethod
+    def _include_change(change: octodns.record.change.Change) -> bool:
+        """filter out record types which the provider can't create in netbox
+        @param change: the planned change
+
+        @return: false if the change should be discarded, true if it should be kept.
+        """
+        if change.record._type in ["SOA", "PTR", "NS"]:
             return False
 
         return True
 
     def _apply(self, plan: octodns.provider.plan.Plan) -> None:
-        """Apply the changes to the NetBox DNS zone."""
+        """apply the changes to the NetBox DNS zone."""
-        self.log.debug(f"_apply: zone={plan.desired.name}, len(changes)={len(plan.changes)}")
+        self.log.debug(f"_apply: zone={plan.desired.name}, changes={len(plan.changes)}")
 
         nb_zone = self._get_nb_zone(plan.desired.name, view=self.nb_view)
 
         for change in plan.changes:
             match change:
                 case octodns.record.Create():
-                    name = change.new.name
+                    rcd_name = "@" if change.new.name == "" else change.new.name
-                    if name == "":
-                        name = "@"
 
-                    match change.new:
+                    new_changeset = self._format_changeset(change.new)
-                        case octodns.record.ValueMixin():
+                    for record in new_changeset:
-                            new = {repr(change.new.value)[1:-1]}
+                        nb_record: pynetbox.core.response.Record = (
-                        case octodns.record.ValuesMixin():
+                            self.api.plugins.netbox_dns.records.create(
-                            new = {repr(v)[1:-1] for v in change.new.values}
+                                zone=nb_zone.id,
-                        case _:
+                                name=rcd_name,
-                            raise ValueError
+                                type=change.new._type,
-
+                                ttl=change.new.ttl,
-                    for value in new:
+                                value=record.replace("\\\\", "\\").replace("\\;", ";"),
-                        nb_record = self.api.plugins.netbox_dns.records.create(
+                                disable_ptr=True,
-                            zone=nb_zone.id,
+                            )
-                            name=name,
-                            type=change.new._type,
-                            ttl=change.new.ttl,
-                            value=value.replace("\\\\", "\\").replace("\\;", ";"),
-                            disable_ptr=True,
                         )
                         self.log.debug(f"{nb_record!r}")
 
                 case octodns.record.Delete():
-                    name = change.existing.name
+                    nb_records: pynetbox.core.response.RecordSet = (
-                    if name == "":
+                        self.api.plugins.netbox_dns.records.filter(
-                        name = "@"
+                            zone_id=nb_zone.id,
-
+                            name=change.existing.name,
-                    nb_records = self.api.plugins.netbox_dns.records.filter(
+                            type=change.existing._type,
-                        zone_id=nb_zone.id,
+                        )
-                        name=change.existing.name,
-                        type=change.existing._type,
                     )
 
-                    match change.existing:
+                    existing_changeset = self._format_changeset(change.existing)
-                        case octodns.record.ValueMixin():
-                            existing = {repr(change.existing.value)[1:-1]}
-                        case octodns.record.ValuesMixin():
-                            existing = {repr(v)[1:-1] for v in change.existing.values}
-                        case _:
-                            raise ValueError
-
                     for nb_record in nb_records:
-                        for value in existing:
+                        for record in existing_changeset:
-                            if nb_record.value == value:
+                            if nb_record.value == record:
                                 self.log.debug(
-                                    f"{nb_record.id} {nb_record.name} {nb_record.type} {nb_record.value} {value}"
+                                    f"{nb_record.id} {nb_record.name} {nb_record.type} {nb_record.value} {record}"
                                 )
                                 self.log.debug(f"{nb_record.url} {nb_record.endpoint.url}")
                                 nb_record.delete()
 
                 case octodns.record.Update():
-                    name = change.existing.name
+                    rcd_name = "@" if change.existing.name == "" else change.existing.name
-                    if name == "":
-                        name = "@"
 
-                    nb_records = self.api.plugins.netbox_dns.records.filter(
+                    nb_records: pynetbox.core.response.RecordSet = (
-                        zone_id=nb_zone.id,
+                        self.api.plugins.netbox_dns.records.filter(
-                        name=name,
+                            zone_id=nb_zone.id,
-                        type=change.existing._type,
+                            name=rcd_name,
+                            type=change.existing._type,
+                        )
                     )
 
-                    match change.existing:
+                    existing_changeset = self._format_changeset(change.existing)
-                        case octodns.record.ValueMixin():
+                    new_changeset = self._format_changeset(change.new)
-                            existing = {repr(change.existing.value)[1:-1]}
-                        case octodns.record.ValuesMixin():
-                            existing = {repr(v)[1:-1] for v in change.existing.values}
-                        case _:
-                            raise ValueError
 
-                    match change.new:
+                    to_delete = existing_changeset.difference(new_changeset)
-                        case octodns.record.ValueMixin():
+                    to_update = existing_changeset.intersection(new_changeset)
-                            new = {repr(change.new.value)[1:-1]}
+                    to_create = new_changeset.difference(existing_changeset)
-                        case octodns.record.ValuesMixin():
-                            new = {repr(v)[1:-1] for v in change.new.values}
-                        case _:
-                            raise ValueError
-
-                    delete = existing.difference(new)
-                    update = existing.intersection(new)
-                    create = new.difference(existing)
 
                     for nb_record in nb_records:
-                        if nb_record.value in delete:
+                        if nb_record.value in to_delete:
                             nb_record.delete()
-                        if nb_record.value in update:
+                        if nb_record.value in to_update:
                             nb_record.ttl = change.new.ttl
                             nb_record.save()
 
-                    for value in create:
+                    for record in to_create:
-                        nb_record = self.api.plugins.netbox_dns.records.create(
+                        nb_record: pynetbox.core.response.Record = (
-                            zone=nb_zone.id,
+                            self.api.plugins.netbox_dns.records.create(
-                            name=name,
+                                zone=nb_zone.id,
-                            type=change.new._type,
+                                name=rcd_name,
-                            ttl=change.new.ttl,
+                                type=change.new._type,
-                            value=value.replace("\\\\", "\\").replace("\\;", ";"),
+                                ttl=change.new.ttl,
-                            disable_ptr=True,
+                                value=record.replace("\\\\", "\\").replace("\\;", ";"),
+                                disable_ptr=True,
+                            )
                         )
                         self.log.debug(f"{nb_record!r}")