Bump version 0.2.1 → 0.3.0

.gitea/workflows/build.yml

@@ -0,0 +1,16 @@
+name: build and publish
+
+on:
+  push:
+    tags:
+      - "v*.*.*"
+
+  pull_request:
+    branches: [main, master]
+
+jobs:
+  build-pypackage:
+    uses: actions/workflows/.gitea/workflows/build_pypackage.yml@master
+    secrets:
+      username: __token__
+      token: ${{ secrets.PACKAGE_TOKEN }}

.gitea/workflows/changelog.yml

@@ -0,0 +1,10 @@
+name: update changelog
+
+on:
+  push:
+    tags:
+      - "v*.*.*"
+
+jobs:
+  update-changelog:
+    uses: actions/workflows/.gitea/workflows/update_changelog.yml@master

.gitea/workflows/check_code.yml

@@ -1,4 +1,4 @@
-name: check/lint python code with hatch
+name: check code
 
 on:
   push:
@@ -8,17 +8,7 @@ on:
     branches: [main, master]
 
 jobs:
-  check-python-hatch:
-    runs-on: python311
-    steps:
-      - name: checkout code
-        uses: actions/checkout@v3
-
-      - name: install hatch
-        run: pip install -U hatch
-
-      - name: test codestyle
-        run: hatch run lint:style
-
-      - name: test typing
-        run: hatch run lint:typing
+  check-code:
+    uses: actions/workflows/.gitea/workflows/check_python_hatch.yml@master
+    with:
+      run-tests: false

README.md

@@ -31,6 +31,12 @@ providers:
 
 ## install
 
+### via pip
+
+```bash
+pip install octodns-netbox-dns
+```
+
 ### via pip + git
 
 ```bash

pyproject.toml

@@ -3,7 +3,7 @@ requires = ["hatchling>=1.18", "hatch-regex-commit>=0.0.3"]
 build-backend = "hatchling.build"
 
 [project]
-name = "octodns_netbox_dns"
+name = "octodns-netbox-dns"
 description = "octodns netbox-dns provider"
 readme = "README.md"
 license = "MIT"

src/octodns_netbox_dns/__about__.py

@@ -1 +1 @@
-__version__ = "0.2.1"
+__version__ = "0.3.0"

src/octodns_netbox_dns/__init__.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Literal
+from typing import Any, Literal
 
 import dns.rdata
 import octodns.record
@@ -16,7 +16,7 @@ class NetBoxDNSSource(octodns.source.base.BaseSource):
 
     SUPPORTS_GEO = False
     SUPPORTS_DYNAMIC = False
-    SUPPORTS: set[str] = {
+    SUPPORTS: set[str] = {  # noqa
         "A",
         "AAAA",
         "AFSDB",
@@ -49,7 +49,7 @@ class NetBoxDNSSource(octodns.source.base.BaseSource):
 
     def __init__(
         self,
-        id: int,
+        id: int,  # noqa
         url: str,
         token: str,
         view: str | None | Literal[False] = False,
@@ -63,29 +63,49 @@ class NetBoxDNSSource(octodns.source.base.BaseSource):
         self.log = logging.getLogger(f"NetboxDNSSource[{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 = self._get_view(view)
-        self._ttl = ttl
+
+        self.api = pynetbox.core.api.Api(url, token)
+        self.nb_view = self._get_nb_view(view)
+        self.ttl = ttl
         self.replace_duplicates = replace_duplicates
         self.make_absolute = make_absolute
 
     def _make_absolute(self, value: str) -> str:
-        if not self.make_absolute or value[-1] == ".":
-            return value
-        return value + "."
+        """
+        Return dns name with trailing dot to make it absolute
 
-    def _get_view(self, view: str | None | Literal[False]) -> dict[str, int | str]:
+        @param value: dns record value
+
+        @return: absolute dns record value
+        """
+        if not self.make_absolute or value.endswith("."):
+            return value
+
+        absolute_value = value + "."
+        self.log.debug(f"relative={value}, absolute={absolute_value}")
+
+        return absolute_value
+
+    def _get_nb_view(self, view: str | None | Literal[False]) -> dict[str, int | str]:
+        """
+        Get the correct netbox view when requested
+
+        @param view: `False` for no view, `None` for zones without a view, else the view name
+
+        @return: the netbox view id in the netbox query format
+        """
         if view is False:
             return {}
         if view is None:
             return {"view": "null"}
 
-        nb_view: pynetbox.core.response.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:
-            msg = 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 view={nb_view.name}, id={nb_view.id}")
 
         return {"view_id": nb_view.id}
 
@@ -93,47 +113,29 @@ class NetBoxDNSSource(octodns.source.base.BaseSource):
         """
         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
+
         @raise pynetbox.RequestError: if declared view is not existent
+
+        @return: the netbox dns zone object
         """
         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)
+
+        self.log.debug(f"found zone={nb_zone.name}, id={nb_zone.id}")
 
         return nb_zone
 
-    def populate(self, zone: octodns.zone.Zone, target: bool = False, lenient: bool = False):
+    def _format_rdata(self, rdata: dns.Rdata, raw_value: str) -> dict[str, Any]:
         """
-        Get all the records of a zone from NetBox and add them to the OctoDNS zone.
+        Format netbox record values to correct octodns record values
+
+        @param rdata: rrdata record value
+        @param raw_value: raw record value
+
+        @return: formatted rrdata value
         """
-        self.log.debug(f"populate: name={zone.name}, target={target}, lenient={lenient}")
-
-        records = {}
-
-        nb_zone = self._get_nb_zone(zone.name, view=self._nb_view)
-        if not nb_zone:
-            self.log.error(f"Zone '{zone.name[:-1]}' not found in view: '{self._nb_view}'")
-            raise LookupError
-
-        nb_records = self._api.plugins.netbox_dns.records.filter(zone_id=nb_zone.id)
-        for nb_record in nb_records:
-            self.log.debug(f"{nb_record.name!r} {nb_record.type!r} {nb_record.value!r}")
-
-            rcd_name: str = nb_record.name if nb_record.name != "@" else ""
-            rcd_value: str = nb_record.value if nb_record.value != "@" else nb_record.zone.name
-
-            if nb_record.ttl:
-                nb_ttl = nb_record.ttl
-            elif nb_record.type == "NS":
-                nb_ttl = nb_zone.soa_refresh
-            else:
-                nb_ttl = nb_zone.default_ttl
-
-            data = {
-                "name": rcd_name,
-                "type": nb_record.type,
-                "ttl": nb_ttl,
-                "values": [],
-            }
-            rdata = dns.rdata.from_text("IN", nb_record.type, rcd_value)
         match rdata.rdtype.name:
             case "A" | "AAAA":
                 value = rdata.address
@@ -190,12 +192,8 @@ class NetBoxDNSSource(octodns.source.base.BaseSource):
                     "fingerprint": rdata.fingerprint,
                 }
 
-                case "SOA":
-                    self.log.debug("SOA")
-                    continue
-
             case "SPF" | "TXT":
-                    value = rcd_value.replace(";", r"\;")
+                value = raw_value.replace(";", r"\;")
 
             case "SRV":
                 value = {
@@ -205,16 +203,84 @@ class NetBoxDNSSource(octodns.source.base.BaseSource):
                     "target": self._make_absolute(rdata.target.to_text()),
                 }
 
+            case "SOA":
+                self.log.warning("SOA record type not implemented")
+                raise NotImplementedError
+
             case _:
+                self.log.error("invalid record type")
                 raise ValueError
 
-            if (rcd_name, nb_record.type) not in records:
-                records[(rcd_name, nb_record.type)] = data
-            records[(rcd_name, nb_record.type)]["values"].append(value)
+        self.log.debug(f"formatted record value={value}")
 
-        for data in records.values():
-            if len(data["values"]) == 1:
-                data["value"] = data.pop("values")[0]
+        return value
+
+    def _format_nb_records(self, zone: octodns.zone.Zone) -> list[dict[str, Any]]:
+        """
+        Format netbox dns records to the octodns format
+
+        @param zone: octodns zone
+
+        @return: a list of octodns compatible record dicts
+        """
+        records: dict[tuple[str, str], dict[str, Any]] = {}
+
+        nb_zone = self._get_nb_zone(zone.name, view=self.nb_view)
+        if not nb_zone:
+            self.log.error(f"zone={zone.name}, not found in view={self.nb_view}")
+            raise LookupError
+
+        nb_records: pynetbox.core.response.RecordSet = self.api.plugins.netbox_dns.records.filter(
+            zone_id=nb_zone.id
+        )
+        for nb_record in nb_records:
+            nb_record: pynetbox.core.response.Record
+
+            rcd_name: str = nb_record.name if nb_record.name != "@" else ""
+            raw_value: str = nb_record.value if nb_record.value != "@" else nb_record.zone.name
+            rcd_type: str = nb_record.type
+            rcd_ttl: int = nb_record.ttl or nb_zone.default_ttl
+            if nb_record.type == "NS":
+                rcd_ttl = nb_zone.soa_refresh
+
+            rcd_data = {
+                "name": rcd_name,
+                "type": rcd_type,
+                "ttl": rcd_ttl,
+                "values": [],
+            }
+
+            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)
+            except NotImplementedError:
+                continue
+            except Exception as exc:
+                raise exc
+
+            if (rcd_name, rcd_type) not in records:
+                records[(rcd_name, rcd_type)] = rcd_data
+
+            records[(rcd_name, rcd_type)]["values"].append(rcd_value)
+
+        return list(records.values())
+
+    def populate(
+        self, zone: octodns.zone.Zone, target: bool = False, lenient: bool = False
+    ) -> None:
+        """
+        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.
+        """
+        self.log.info(f"populate -> name={zone.name}, target={target}, lenient={lenient}")
+
+        records = self._format_nb_records(zone)
+        for data in records:
             record = octodns.record.Record.new(
                 zone=zone,
                 name=data["name"],
@@ -224,4 +290,4 @@ class NetBoxDNSSource(octodns.source.base.BaseSource):
             )
             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}")