PENG-7708-python-sdk-zoneexport

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.28.0 (TBD)
+
+ENHANCEMENTS:
+* Add zone export functionality to export zones in BIND format for backup/migration
+
 ## 0.27.1 (December 3rd, 2025)
 
 FIXED:

examples/zone-export.py

@@ -0,0 +1,37 @@
+#
+# Copyright (c) 2026 NSONE, Inc.
+#
+# License under The MIT License (MIT). See LICENSE in project root.
+#
+
+from ns1 import NS1
+
+# NS1 will use config in ~/.nsone by default
+api = NS1()
+
+# to specify an apikey here instead, use:
+# api = NS1(apiKey='<<CLEARTEXT API KEY>>')
+
+# to load an alternate configuration file:
+# api = NS1(configFile='/etc/ns1/api.json')
+
+# Define the zone to export
+zone_name = "example.com"
+
+# Export a zone to BIND format
+# The export() method will:
+# 1. Initiate the export job
+# 2. Poll the status until complete or failed
+# 3. Download and return the zone file content
+zone = api.loadZone(zone_name)
+
+print(f"Exporting zone {zone_name}...")
+zone_file = zone.export()
+print("Export complete!")
+print(zone_file)
+
+# Save to a file
+output_file = f"{zone_name}.txt"
+with open(output_file, "w") as f:
+    f.write(zone_file)
+print(f"Zone file saved to {output_file}")

ns1/__init__.py

@@ -5,7 +5,7 @@
 #
 from .config import Config
 
-version = "0.27.1"
+version = "0.28.0"
 
 
 class NS1:

ns1/rest/zones.py

@@ -5,6 +5,7 @@
 #
 
 from . import resource
+from .errors import ResourceException
 
 
 class Zones(resource.BaseResource):
@@ -188,6 +189,64 @@ def delete_version(self, zone, version_id, callback=None, errback=None):
             errback=errback,
         )
 
+    def initiate_zonefile_export(self, zone, callback=None, errback=None):
+        """
+        Initiate zone export job.
+
+        :param str zone: zone name
+        :return: export status response
+        """
+        return self._make_request(
+            "PUT",
+            f"export/zonefile/{zone}",
+            body={},
+            callback=callback,
+            errback=errback,
+        )
+
+    def status_zonefile_export(self, zone, callback=None, errback=None):
+        """
+        Check zone export status.
+
+        :param str zone: zone name
+        :return: export status response
+        """
+        return self._make_request(
+            "GET",
+            f"export/zonefile/{zone}/status",
+            callback=callback,
+            errback=errback,
+        )
+
+    def get_zonefile_export(self, zone, callback=None, errback=None):
+        """
+        Download the exported zone file in BIND-compatible format.
+
+        :param str zone: zone name
+        :return: zone file content as string
+        """
+        # Note: This endpoint returns raw zone file text, not JSON
+        # The transport layer will try to parse it as JSON and fail
+        # We catch that exception and extract the raw body text
+        try:
+            return self._make_request(
+                "GET",
+                f"export/zonefile/{zone}",
+                callback=callback,
+                errback=errback,
+            )
+        except ResourceException as e:
+            # Check if this is a valid zonefile response (plain text)
+            # The response should be 200 OK with text/plain content
+            if e.response and e.response.getcode() == 200:
+                # Check content-type header for text/plain
+                content_type = e.response.getheader("Content-Type", "")
+                if "text/plain" in content_type or "text" in content_type:
+                    # This is the expected plain text zonefile
+                    return e.body if e.body else ""
+            # Otherwise, this is a real error - re-raise it
+            raise
+
 
 # successive pages just extend the list of zones
 def zone_list_pagination(curr_json, next_json):

ns1/zones.py

@@ -3,6 +3,8 @@
 #
 # License under The MIT License (MIT). See LICENSE in project root.
 #
+import time
+
 from ns1.rest.zones import Zones
 from ns1.records import Record
 from ns1.rest.stats import Stats
@@ -123,15 +125,15 @@ def success(result, *args):
                 callback=success,
                 errback=errback,
                 name=name,
-                **kwargs
+                **kwargs,
             )
         else:
             return self._rest.create(
                 self.zone,
                 callback=success,
                 errback=errback,
                 name=name,
-                **kwargs
+                **kwargs,
             )
 
     def __getattr__(self, item):
@@ -169,7 +171,7 @@ def linkRecord(
         rtype,
         callback=None,
         errback=None,
-        **kwargs
+        **kwargs,
     ):
         """
         Create a new linked record in this zone. These records use the
@@ -194,7 +196,7 @@ def linkRecord(
             link=existing_domain,
             callback=callback,
             errback=errback,
-            **kwargs
+            **kwargs,
         )
 
     def cloneRecord(
@@ -287,3 +289,51 @@ def usage(self, callback=None, errback=None, **kwargs):
         return stats.usage(
             zone=self.zone, callback=callback, errback=errback, **kwargs
         )
+
+    def export(
+        self, callback=None, errback=None, timeout=300, poll_interval=2
+    ):
+        """
+        Export zone as a BIND-compatible zone file.
+
+        This method initiates the export, polls the status until complete or failed,
+        and downloads the zone file.
+
+        :param callback: optional callback
+        :param errback: optional error callback
+        :param int timeout: maximum time to wait for export completion in seconds (default: 300)
+        :param int poll_interval: time between status checks in seconds (default: 2)
+        :return: zone file content as string
+        :raises ZoneException: if export fails or times out
+        """
+        # Initiate the export
+        init_response = self._rest.initiate_zonefile_export(self.zone)
+        if not init_response or init_response.get("status") == "FAILED":
+            error_msg = init_response.get(
+                "message", "Failed to initiate export"
+            )
+            raise ZoneException(f"Zone export initiation failed: {error_msg}")
+
+        # Poll the status until complete or failed
+        start_time = time.time()
+        while True:
+            if time.time() - start_time > timeout:
+                raise ZoneException(
+                    f"Zone export timed out after {timeout} seconds"
+                )
+
+            status_response = self._rest.status_zonefile_export(self.zone)
+            status = status_response.get("status")
+
+            if status == "COMPLETED":
+                break
+            elif status == "FAILED":
+                error_msg = status_response.get("message", "Unknown error")
+                raise ZoneException(f"Zone export failed: {error_msg}")
+
+            time.sleep(poll_interval)
+
+        # Download the zone file
+        return self._rest.get_zonefile_export(
+            self.zone, callback=callback, errback=errback
+        )

tests/unit/test_zone.py

@@ -251,3 +251,49 @@ def test_rest_zone_buildbody(zones_config):
         "tags": {"foo": "bar", "hai": "bai"},
     }
     assert z._buildBody(zone, **kwargs) == body
+
+
+@pytest.mark.parametrize(
+    "zone, url", [("test.zone", "export/zonefile/test.zone")]
+)
+def test_rest_zone_get_zonefile_export(zones_config, zone, url):
+    z = ns1.rest.zones.Zones(zones_config)
+    z._make_request = mock.MagicMock()
+    z.get_zonefile_export(zone)
+    z._make_request.assert_called_once_with(
+        "GET",
+        url,
+        callback=None,
+        errback=None,
+    )
+
+
+@pytest.mark.parametrize(
+    "zone, url", [("test.zone", "export/zonefile/test.zone")]
+)
+def test_rest_zone_initiate_zonefile_export(zones_config, zone, url):
+    z = ns1.rest.zones.Zones(zones_config)
+    z._make_request = mock.MagicMock()
+    z.initiate_zonefile_export(zone)
+    z._make_request.assert_called_once_with(
+        "PUT",
+        url,
+        body={},
+        callback=None,
+        errback=None,
+    )
+
+
+@pytest.mark.parametrize(
+    "zone, url", [("test.zone", "export/zonefile/test.zone/status")]
+)
+def test_rest_zone_status_zonefile_export(zones_config, zone, url):
+    z = ns1.rest.zones.Zones(zones_config)
+    z._make_request = mock.MagicMock()
+    z.status_zonefile_export(zone)
+    z._make_request.assert_called_once_with(
+        "GET",
+        url,
+        callback=None,
+        errback=None,
+    )