diff --git a/src/octodns_netbox_dns/__init__.py b/src/octodns_netbox_dns/__init__.py index 6d2e980..d7891fe 100644 --- a/src/octodns_netbox_dns/__init__.py +++ b/src/octodns_netbox_dns/__init__.py @@ -1,20 +1,22 @@ -"""OctoDNS provider for NetboxDNS.""" import logging from typing import Literal import dns.rdata -import octodns.provider.base -import octodns.provider.plan import octodns.record +import octodns.source.base import octodns.zone import pynetbox.core.api -import pynetbox.core.response as pynb_resp +import pynetbox.core.response -class NetBoxDNSSource(octodns.provider.base.BaseProvider): - SUPPORTS_GEO: bool = False - SUPPORTS_DYNAMIC: bool = False - SUPPORTS: set[str] = { +class NetBoxDNSSource(octodns.source.base.BaseSource): + """ + OctoDNS provider for NetboxDNS + """ + + SUPPORTS_GEO = False + SUPPORTS_DYNAMIC = False + SUPPORTS: frozenset = { "A", "AAAA", "AFSDB", @@ -45,10 +47,6 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider): "TXT", } - _api: pynetbox.core.api.Api - # log: logging.Logger - _ttl: int - def __init__( self, id: int, @@ -56,15 +54,15 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider): token: str, view: str | None | Literal[False] = False, ttl=3600, - replace_duplicates: bool = False, - make_absolute: bool = False, + replace_duplicates=False, + make_absolute=False, ): - """Initialize the NetboxDNSSource.""" + """ + Initialize the NetboxDNSSource + """ self.log = logging.getLogger(f"NetboxDNSSource[{id}]") - self.log.debug( - f"__init__: id={id}, url={url}, view={view}, replace_duplicates={replace_duplicates}, make_absolute={make_absolute}" - ) - super(NetBoxDNSSource, self).__init__(id) + self.log.debug(f"__init__: {id=}, {url=}, {view=}, {replace_duplicates=}, {make_absolute=}") + super().__init__(id) self._api = pynetbox.core.api.Api(url, token) self._nb_view = {} if view is False else self._get_view(view) self._ttl = ttl @@ -76,27 +74,34 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider): return value return value + "." - def _get_view(self, view: str | None) -> dict[str, int | str]: + def _get_view(self, view: str | Literal[False]) -> dict[str, int | str]: if view is None: return {"view": "null"} - nb_view: pynb_resp.Record = self._api.plugins.netbox_dns.views.get(name=view) + nb_view: pynetbox.core.response.Record = self._api.plugins.netbox_dns.views.get(name=view) if nb_view is None: - raise ValueError(f"dns view: '{view}' has not been found") + msg = f"dns view: '{view}' has not been found" + self.log.error(msg) + raise ValueError(msg) self.log.debug(f"found {nb_view.name} {nb_view.id}") return {"view_id": nb_view.id} - def _get_nb_zone(self, name: str, view: dict[str, str | int]) -> pynb_resp.Record: - """Given a zone name and a view name, look it up in NetBox. - Raises: pynetbox.RequestError if declared view is not existant""" + 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. + + @raise pynetbox.RequestError: if declared view is not existent + """ query_params = {"name": name[:-1], **view} nb_zone = self._api.plugins.netbox_dns.zones.get(**query_params) return nb_zone def populate(self, zone: octodns.zone.Zone, target: bool = False, lenient: bool = False): - """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. + """ self.log.debug(f"populate: name={zone.name}, target={target}, lenient={lenient}") records = {} @@ -218,111 +223,3 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider): zone.add_record(record, lenient=lenient, replace=self.replace_duplicates) self.log.info(f"populate: found {len(zone.records)} records for zone {zone.name}") - - def _apply(self, plan: octodns.provider.plan.Plan): - """Apply the changes to the NetBox DNS zone.""" - self.log.debug(f"_apply: zone={plan.desired.name}, len(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 - if name == "": - name = "@" - - match change.new: - case octodns.record.ValueMixin(): - new = {repr(change.new.value)[1:-1]} - case octodns.record.ValuesMixin(): - new = set(map(lambda v: repr(v)[1:-1], change.new.values)) - case _: - raise ValueError - - for value in new: - nb_record = self._api.plugins.netbox_dns.records.create( - zone=nb_zone.id, - name=name, - type=change.new._type, - ttl=change.new.ttl, - value=value, - disable_ptr=True, - ) - self.log.debug(f"{nb_record!r}") - - case octodns.record.Delete(): - name = change.existing.name - if name == "": - name = "@" - - nb_records = self._api.plugins.netbox_dns.records.filter( - zone_id=nb_zone.id, - name=change.existing.name, - type=change.existing._type, - ) - - match change.existing: - case octodns.record.ValueMixin(): - existing = {repr(change.existing.value)[1:-1]} - case octodns.record.ValuesMixin(): - existing = set(map(lambda v: repr(v)[1:-1], change.existing.values)) - case _: - raise ValueError - - for nb_record in nb_records: - for value in existing: - if nb_record.value == value: - self.log.debug( - f"{nb_record.id} {nb_record.name} {nb_record.type} {nb_record.value} {value}" - ) - self.log.debug(f"{nb_record.url} {nb_record.endpoint.url}") - nb_record.delete() - - case octodns.record.Update(): - name = change.existing.name - if name == "": - name = "@" - - nb_records = self._api.plugins.netbox_dns.records.filter( - zone_id=nb_zone.id, - name=name, - type=change.existing._type, - ) - - match change.existing: - case octodns.record.ValueMixin(): - existing = {repr(change.existing.value)[1:-1]} - case octodns.record.ValuesMixin(): - existing = set(map(lambda v: repr(v)[1:-1], change.existing.values)) - case _: - raise ValueError - - match change.new: - case octodns.record.ValueMixin(): - new = {repr(change.new.value)[1:-1]} - case octodns.record.ValuesMixin(): - new = set(map(lambda v: repr(v)[1:-1], 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: - nb_record.delete() - if nb_record.value in update: - nb_record.ttl = change.new.ttl - nb_record.save() - - for value in create: - nb_record = self._api.plugins.netbox_dns.records.create( - zone=nb_zone.id, - name=name, - type=change.new._type, - ttl=change.new.ttl, - value=value, - disable_ptr=True, - )