olofvndrhr
/
octodns-netbox-dns
1
0
Fork 0
Code Issues 1 Pull Requests 9 Actions Packages Projects Releases 4 Wiki Activity

move from provider base class to source

This commit is contained in:
Ivan Schaller 2024-01-09 22:54:52 +01:00
parent 4a2ba28aaa
commit 5714ce5af9
1 changed files with 31 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

165
src/octodns_netbox_dns/__init__.py
View File

@ -1,20 +1,22 @@
"""OctoDNS provider for NetboxDNS."""
import logging import logging
from typing import Literal from typing import Literal
import dns.rdata import dns.rdata
import octodns.provider.base
import octodns.provider.plan
import octodns.record import octodns.record
import octodns.source.base
import octodns.zone import octodns.zone
import pynetbox.core.api import pynetbox.core.api
import pynetbox.core.response as pynb_resp import pynetbox.core.response
class NetBoxDNSSource(octodns.provider.base.BaseProvider): class NetBoxDNSSource(octodns.source.base.BaseSource):
SUPPORTS_GEO: bool = False """
SUPPORTS_DYNAMIC: bool = False OctoDNS provider for NetboxDNS
SUPPORTS: set[str] = { """
SUPPORTS_GEO = False
SUPPORTS_DYNAMIC = False
SUPPORTS: frozenset = {
"A", "A",
"AAAA", "AAAA",
"AFSDB", "AFSDB",
@ -45,10 +47,6 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
"TXT", "TXT",
} }
_api: pynetbox.core.api.Api
# log: logging.Logger
_ttl: int
def __init__( def __init__(
self, self,
id: int, id: int,
@ -56,15 +54,15 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
token: str, token: str,
view: str | None | Literal[False] = False, view: str | None | Literal[False] = False,
ttl=3600, ttl=3600,
replace_duplicates: bool = False, replace_duplicates=False,
make_absolute: bool = False, make_absolute=False,
): ):
"""Initialize the NetboxDNSSource.""" """
Initialize the NetboxDNSSource
"""
self.log = logging.getLogger(f"NetboxDNSSource[{id}]") self.log = logging.getLogger(f"NetboxDNSSource[{id}]")
self.log.debug( self.log.debug(f"__init__: {id=}, {url=}, {view=}, {replace_duplicates=}, {make_absolute=}")
f"__init__: id={id}, url={url}, view={view}, replace_duplicates={replace_duplicates}, make_absolute={make_absolute}" super().__init__(id)
)
super(NetBoxDNSSource, self).__init__(id)
self._api = pynetbox.core.api.Api(url, token) self._api = pynetbox.core.api.Api(url, token)
self._nb_view = {} if view is False else self._get_view(view) self._nb_view = {} if view is False else self._get_view(view)
self._ttl = ttl self._ttl = ttl
@ -76,27 +74,34 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
return value return value
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: if view is None:
return {"view": "null"} 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: 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}") self.log.debug(f"found {nb_view.name} {nb_view.id}")
return {"view_id": nb_view.id} return {"view_id": nb_view.id}
def _get_nb_zone(self, name: str, view: dict[str, str | int]) -> pynb_resp.Record: 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. """
Raises: pynetbox.RequestError if declared view is not existant""" 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} query_params = {"name": name[:-1], **view}
nb_zone = self._api.plugins.netbox_dns.zones.get(**query_params) nb_zone = self._api.plugins.netbox_dns.zones.get(**query_params)
return nb_zone return nb_zone
def populate(self, zone: octodns.zone.Zone, target: bool = False, lenient: bool = False): 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}") self.log.debug(f"populate: name={zone.name}, target={target}, lenient={lenient}")
records = {} records = {}
@ -218,111 +223,3 @@ class NetBoxDNSSource(octodns.provider.base.BaseProvider):
zone.add_record(record, lenient=lenient, replace=self.replace_duplicates) zone.add_record(record, lenient=lenient, replace=self.replace_duplicates)
self.log.info(f"populate: found {len(zone.records)} records for zone {zone.name}") 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,
)