Add support for batch change zone records endpoint (#477)

lokst · web-flow · commit c820e0445e1b · 2025-09-19T12:36:58.000+08:00
Add support for batch change zone records endpoint Addresses dnsimple/dnsimple-app#31926 Belongs to dnsimple/dnsimple-business#2263
diff --git a/dnsimple/service/zones.py b/dnsimple/service/zones.py
@@ -1,5 +1,5 @@
 from dnsimple.response import Response
-from dnsimple.struct import Zone, ZoneDistribution, ZoneFile, ZoneRecord
+from dnsimple.struct import Zone, ZoneDistribution, ZoneFile, ZoneRecord, BatchChangeZoneRecordsResponse
 
 
 class Zones(object):
@@ -264,3 +264,22 @@ def check_zone_record_distribution(self, account_id, zone, record_id):
         """
         response = self.client.get(f'/{account_id}/zones/{zone}/records/{record_id}/distribution')
         return Response(response, ZoneDistribution)
+
+    def batch_change_records(self, account_id, zone, batch_change):
+        """
+        Batch change zone records in the account.
+
+        See https://developer.dnsimple.com/v2/zones/records/#batchChangeZoneRecords
+
+        :param account_id: int
+            The account ID
+        :param zone: str
+            The zone name
+        :param batch_change: dnsimple.struct.BatchChangeZoneRecordsInput
+            The data to send to batch change zone records
+
+        :return: dnsimple.Response
+            The batch change zone records response
+        """
+        response = self.client.post(f'/{account_id}/zones/{zone}/batch', data=batch_change.to_json())
+        return Response(response, BatchChangeZoneRecordsResponse)
diff --git a/dnsimple/struct/__init__.py b/dnsimple/struct/__init__.py
@@ -33,3 +33,11 @@
 from dnsimple.struct.zone_distribution import ZoneDistribution
 from dnsimple.struct.zone_file import ZoneFile
 from dnsimple.struct.zone_record import ZoneRecord, ZoneRecordInput, ZoneRecordUpdateInput
+from dnsimple.struct.batch_change_zone_records import (
+    BatchChangeZoneRecordsInput,
+    BatchChangeZoneRecordsCreateInput,
+    BatchChangeZoneRecordsUpdateInput,
+    BatchChangeZoneRecordsDeleteInput,
+    BatchChangeZoneRecordsResponse,
+    BatchChangeZoneRecordsDeleteResponse
+)
diff --git a/dnsimple/struct/batch_change_zone_records.py b/dnsimple/struct/batch_change_zone_records.py
@@ -0,0 +1,100 @@
+import json
+from dataclasses import dataclass
+
+import omitempty
+
+from dnsimple.struct import Struct
+from dnsimple.struct.zone_record import ZoneRecord, ZoneRecordInput
+
+BatchChangeZoneRecordsCreateInput = ZoneRecordInput
+
+
+@dataclass
+class BatchChangeZoneRecordsUpdateInput(dict):
+    """Represents a zone record update input for a batch operation"""
+
+    def __init__(self, id, name=None, content=None, ttl=None, priority=None, regions=None):
+        dict.__init__(self, id=id, name=name, content=content, ttl=ttl, priority=priority, regions=regions)
+
+    def to_json(self):
+        omitted = omitempty(self)
+
+        if self['name'] == '':
+            omitted['name'] = ''
+
+        return json.dumps(omitted)
+
+
+@dataclass
+class BatchChangeZoneRecordsDeleteInput(dict):
+    """Represents a zone record deletion input for a batch operation"""
+
+    def __init__(self, id):
+        dict.__init__(self, id=id)
+
+
+@dataclass
+class BatchChangeZoneRecordsInput(dict):
+    """Represents the data to send to the DNSimple API to make a batch change on the records of a zone
+
+    All parameters are optional - you can perform creates only, updates only, deletes only,
+    or any combination of the three operations.
+
+    :param creates: List[BatchChangeZoneRecordsCreateInput] - Records to create (optional)
+    :param updates: List[BatchChangeZoneRecordsUpdateInput] - Records to update (optional)
+    :param deletes: List[BatchChangeZoneRecordsDeleteInput] - Records to delete (optional)
+    """
+
+    def __init__(self, creates=None, updates=None, deletes=None):
+        data = {}
+        if creates is not None:
+            data['creates'] = creates
+        if updates is not None:
+            data['updates'] = updates
+        if deletes is not None:
+            data['deletes'] = deletes
+        dict.__init__(self, **data)
+
+    def to_json(self):
+        result = {}
+
+        if 'creates' in self and self['creates'] is not None:
+            result['creates'] = [json.loads(item.to_json()) for item in self['creates']]
+
+        if 'updates' in self and self['updates'] is not None:
+            result['updates'] = [json.loads(item.to_json()) for item in self['updates']]
+
+        if 'deletes' in self and self['deletes'] is not None:
+            result['deletes'] = [omitempty(item) for item in self['deletes']]
+
+        return json.dumps(result)
+
+
+@dataclass
+class BatchChangeZoneRecordsDeleteResponse(Struct):
+    """Represents a deleted zone record in the batch change response"""
+    id = None
+    """The record ID that was deleted"""
+
+    def __init__(self, data):
+        super().__init__(data)
+
+
+@dataclass
+class BatchChangeZoneRecordsResponse(Struct):
+    """Represents the response from batch changing zone records"""
+    creates = None
+    """List of created zone records"""
+    updates = None
+    """List of updated zone records"""
+    deletes = None
+    """List of deleted zone record IDs"""
+
+    def __init__(self, data):
+        super().__init__(data)
+        if 'creates' in data and data['creates'] is not None:
+            self.creates = [ZoneRecord(item) for item in data['creates']]
+        if 'updates' in data and data['updates'] is not None:
+            self.updates = [ZoneRecord(item) for item in data['updates']]
+        if 'deletes' in data and data['deletes'] is not None:
+            self.deletes = [BatchChangeZoneRecordsDeleteResponse(item) for item in data['deletes']]
diff --git a/tests/service/zones_test.py b/tests/service/zones_test.py
@@ -4,6 +4,14 @@
 
 from dnsimple import DNSimpleException
 from dnsimple.struct.zone import Zone
+from dnsimple.struct.batch_change_zone_records import (
+    BatchChangeZoneRecordsInput,
+    BatchChangeZoneRecordsUpdateInput,
+    BatchChangeZoneRecordsDeleteInput,
+    BatchChangeZoneRecordsResponse
+)
+from dnsimple.struct.zone_record import ZoneRecordInput
+from dnsimple.struct.zone_record import ZoneRecord
 from tests.helpers import DNSimpleTest, DNSimpleMockResponse
 
 
@@ -126,6 +134,105 @@ def test_check_zone_distribution_failure(self):
             self.assertEqual('Could not query zone, connection time out', dnse.message)
             self.assertIsInstance(dnse, DNSimpleException)
 
+    @responses.activate
+    def test_batch_change_records_success(self):
+        responses.add(DNSimpleMockResponse(method=responses.POST,
+                                           path='/1010/zones/example.com/batch',
+                                           fixture_name='batchChangeZoneRecords/success'))
+
+        batch_change = BatchChangeZoneRecordsInput(
+            creates=[
+                ZoneRecordInput('ab', 'A', '3.2.3.4'),
+                ZoneRecordInput('ab', 'A', '4.2.3.4')
+            ],
+            updates=[
+                BatchChangeZoneRecordsUpdateInput(67622534, content='3.2.3.40'),
+                BatchChangeZoneRecordsUpdateInput(67622537, content='5.2.3.40')
+            ],
+            deletes=[
+                BatchChangeZoneRecordsDeleteInput(67622509),
+                BatchChangeZoneRecordsDeleteInput(67622527)
+            ]
+        )
+
+        response = self.zones.batch_change_records(1010, 'example.com', batch_change)
+        result = response.data
+
+        self.assertIsInstance(result, BatchChangeZoneRecordsResponse)
+
+        self.assertEqual(2, len(result.creates))
+        self.assertIsInstance(result.creates[0], ZoneRecord)
+        self.assertEqual(67623409, result.creates[0].id)
+        self.assertEqual('ab', result.creates[0].name)
+        self.assertEqual('3.2.3.4', result.creates[0].content)
+        self.assertEqual('A', result.creates[0].type)
+
+        self.assertEqual(2, len(result.updates))
+        self.assertIsInstance(result.updates[0], ZoneRecord)
+        self.assertEqual(67622534, result.updates[0].id)
+        self.assertEqual('3.2.3.40', result.updates[0].content)
+
+        self.assertEqual(2, len(result.deletes))
+        self.assertEqual(67622509, result.deletes[0].id)
+        self.assertEqual(67622527, result.deletes[1].id)
+
+    @responses.activate
+    def test_batch_change_records_create_validation_failed(self):
+        responses.add(DNSimpleMockResponse(method=responses.POST,
+                                           path='/1010/zones/example.com/batch',
+                                           fixture_name='batchChangeZoneRecords/error_400_create_validation_failed'))
+
+        batch_change = BatchChangeZoneRecordsInput(
+            creates=[
+                ZoneRecordInput('test', 'SPF', 'v=spf1 -all')
+            ]
+        )
+
+        try:
+            self.zones.batch_change_records(1010, 'example.com', batch_change)
+        except DNSimpleException as dnse:
+            self.assertEqual('Validation failed', dnse.message)
+            self.assertIsInstance(dnse, DNSimpleException)
+            self.assertEqual('The SPF record type has been discontinued', dnse.attribute_errors['creates'][0]['message'])
+
+    @responses.activate
+    def test_batch_change_records_update_validation_failed(self):
+        responses.add(DNSimpleMockResponse(method=responses.POST,
+                                           path='/1010/zones/example.com/batch',
+                                           fixture_name='batchChangeZoneRecords/error_400_update_validation_failed'))
+
+        batch_change = BatchChangeZoneRecordsInput(
+            updates=[
+                BatchChangeZoneRecordsUpdateInput(99999999, content='1.2.3.4')
+            ]
+        )
+
+        try:
+            self.zones.batch_change_records(1010, 'example.com', batch_change)
+        except DNSimpleException as dnse:
+            self.assertEqual('Validation failed', dnse.message)
+            self.assertIsInstance(dnse, DNSimpleException)
+            self.assertEqual('Record not found ID=99999999', dnse.attribute_errors['updates'][0]['message'])
+
+    @responses.activate
+    def test_batch_change_records_delete_validation_failed(self):
+        responses.add(DNSimpleMockResponse(method=responses.POST,
+                                           path='/1010/zones/example.com/batch',
+                                           fixture_name='batchChangeZoneRecords/error_400_delete_validation_failed'))
+
+        batch_change = BatchChangeZoneRecordsInput(
+            deletes=[
+                BatchChangeZoneRecordsDeleteInput(67622509)
+            ]
+        )
+
+        try:
+            self.zones.batch_change_records(1010, 'example.com', batch_change)
+        except DNSimpleException as dnse:
+            self.assertEqual('Validation failed', dnse.message)
+            self.assertIsInstance(dnse, DNSimpleException)
+            self.assertEqual('Record not found ID=67622509', dnse.attribute_errors['deletes'][0]['message'])
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/struct/batch_change_zone_records_test.py b/tests/struct/batch_change_zone_records_test.py
