[AAP-82119] Batch resource migration using bulk-update endpoint

Refactor migrate_resource() to process entire API pages as batches
instead of one item at a time. The new _process_resource_page_batch()
method:

1. Validates all items on the page (reuses existing helpers)
2. Reconciles against existing gateway resources
3. Bulk-creates gateway resources in a single transaction
4. Sends a single bulk_update_resources() HTTP call per page

This reduces HTTP round-trips from N (one per resource) to N/page_size
(one per page). For Kyndryl-scale data (3,962 users), this drops from
3,962 sequential PATCH calls to ~80 batch calls — projected to cut
user migration from 203s to under 10s.

Depends on: django-ansible-base AAP-82119/bulk-update-endpoint branch

Assisted-by: Claude AI assistant

Author: Dostonbek1

aap_gateway_api/management/commands/_migrate_service_data/resource_migration.py

@@ -253,6 +253,72 @@ def _process_and_migrate_resource_item(self, upstream_resource_item: Dict[str, A
 
             self.client.update_resource(resource_ansible_id, ResourceRequestBody(**updated_service_resource), partial=True)
 
+    def _process_resource_page_batch(self, results: List[Dict[str, Any]], resource_context: Dict[str, Any]) -> int:
+        """
+        Process and migrate a batch of resource items from a single API page.
+
+        This replaces the per-item loop with a batch approach that:
+        1. Validates and prepares all items on the page
+        2. Creates gateway resources within a single transaction
+        3. Sends a single bulk HTTP call to update all upstream resources
+
+        Args:
+            results: List of resource items from the upstream service API page
+            resource_context: Static data about the resource type
+
+        Returns:
+            Number of items successfully processed in this batch
+        """
+        resource_type = resource_context["type"]
+        bulk_update_items = []
+        create_operations = []
+
+        for upstream_resource_item in results:
+            resource_ansible_id = upstream_resource_item["ansible_id"]
+
+            if "resource_data" not in upstream_resource_item:
+                raise RuntimeError(
+                    f"Resource {resource_ansible_id} is missing 'resource_data'. Ensure all services are running a version of DAB that supports extra_fields."
+                )
+
+            upstream_resource = upstream_resource_item
+
+            validated_resource_data = self._deserialize_and_validate_resource_data(upstream_resource, resource_context["type_serializer"])
+
+            if resource_context["type_name"] == SHARED_USER_RESOURCE_TYPE:
+                upstream_resource = self._sync_user_superuser_flag(upstream_resource, validated_resource_data)
+                validated_resource_data = self._deserialize_and_validate_resource_data(upstream_resource, resource_context["type_serializer"])
+
+            resource_creation_kwargs, updated_service_resource = self._initialize_resource_sync_payloads(upstream_resource)
+            create_gateway_resource = self._reconcile_existing_resource(upstream_resource, resource_context, validated_resource_data, updated_service_resource)
+
+            if create_gateway_resource:
+                create_operations.append((resource_type, upstream_resource["resource_data"], resource_creation_kwargs))
+
+            # Build the bulk-update payload item for the upstream service
+            bulk_item: Dict[str, Any] = {"ansible_id": resource_ansible_id}
+            if "service_id" in updated_service_resource:
+                bulk_item["service_id"] = updated_service_resource["service_id"]
+            if "is_partially_migrated" in updated_service_resource:
+                bulk_item["is_partially_migrated"] = updated_service_resource["is_partially_migrated"]
+            if "ansible_id" in updated_service_resource:
+                bulk_item["new_ansible_id"] = str(updated_service_resource["ansible_id"])
+            if "resource_data" in updated_service_resource:
+                bulk_item["resource_data"] = updated_service_resource["resource_data"]
+
+            bulk_update_items.append(bulk_item)
+
+        with transaction.atomic():
+            for rt, resource_data, creation_kwargs in create_operations:
+                Resource.create_resource(rt, resource_data, **creation_kwargs)
+
+            if bulk_update_items:
+                resp = self.client.bulk_update_resources(bulk_update_items)
+                if resp.status_code != 200:
+                    raise RuntimeError(f"Bulk resource update failed with status {resp.status_code}: {resp.text}")
+
+        return len(results)
+
     def migrate_resource(self, resource_type_name: str) -> None:
         """
         Migrate all resources of a specific type from upstream service to Gateway.
@@ -328,7 +394,6 @@ def migrate_resource(self, resource_type_name: str) -> None:
                 self._log("No more items remaining to migrate.", logging.INFO)
                 break
 
-            for upstream_resource_item in results:
-                resource_processed += 1
-                self._log_progress(progress_label, resource_processed, resource_total)
-                self._process_and_migrate_resource_item(upstream_resource_item, resource_context)
+            batch_size = self._process_resource_page_batch(results, resource_context)
+            resource_processed += batch_size
+            self._log_progress(progress_label, resource_processed, resource_total)

aap_gateway_api/tests/management/commands/_migrate_service_data/test_resource_migration.py

@@ -175,6 +175,179 @@ def test_process_migrate_resource_item_raises_on_missing_resource_data():
         cmd._process_and_migrate_resource_item(resource_item, resource_context)
 
 
+@pytest.mark.django_db
+def test_process_resource_page_batch_raises_on_missing_resource_data():
+    """Test that _process_resource_page_batch raises when resource_data is missing."""
+    cmd = MigrateCommand()
+    results = [{"ansible_id": "test-id-456", "name": "test"}]
+    resource_context = {"type": Mock()}
+
+    with pytest.raises(RuntimeError, match="missing 'resource_data'"):
+        cmd._process_resource_page_batch(results, resource_context)
+
+
+@pytest.mark.django_db
+def test_process_resource_page_batch_bulk_update():
+    """Test that _process_resource_page_batch calls bulk_update_resources with correct payloads."""
+    import uuid
+
+    from ansible_base.resource_registry.models import ResourceType
+
+    cmd = MigrateCommand()
+    cmd.stdout = StringIO()
+    cmd.stderr = StringIO()
+    cmd.upstream_service_id = str(uuid.uuid4())
+
+    mock_client = Mock()
+    mock_client.service.service_cluster.service_type.name = "awx"
+    mock_resp = Mock()
+    mock_resp.status_code = 200
+    mock_resp.json.return_value = {"updated": 2, "errors": []}
+    mock_client.bulk_update_resources.return_value = mock_resp
+    cmd.client = mock_client
+
+    Organization.objects.create(name="BatchOrg1")
+
+    org_resource_type = ResourceType.objects.get(name="shared.organization")
+    resource_context = {
+        "type": org_resource_type,
+        "type_name": "shared.organization",
+        "type_serializer": org_resource_type.serializer_class,
+        "type_name_field": org_resource_type.get_resource_config().name_field,
+        "unique_fields": ["name"],
+        "LocalResourceModel": Organization,
+    }
+
+    results = [
+        {
+            "ansible_id": str(uuid.uuid4()),
+            "name": "BatchOrg1",
+            "resource_type": "shared.organization",
+            "resource_data": {"name": "BatchOrg1"},
+        },
+        {
+            "ansible_id": str(uuid.uuid4()),
+            "name": "NewOrg",
+            "resource_type": "shared.organization",
+            "resource_data": {"name": "NewOrg"},
+        },
+    ]
+
+    count = cmd._process_resource_page_batch(results, resource_context)
+    assert count == 2
+
+    mock_client.bulk_update_resources.assert_called_once()
+    bulk_items = mock_client.bulk_update_resources.call_args[0][0]
+    assert len(bulk_items) == 2
+    assert all("ansible_id" in item for item in bulk_items)
+    # The first item (BatchOrg1 exists) triggers reconcile which sets ansible_id and resource_data
+    merged_item = bulk_items[0]
+    assert "new_ansible_id" in merged_item
+    assert "resource_data" in merged_item
+    # The second item (NewOrg is new) only gets service_id
+    new_item = bulk_items[1]
+    assert "service_id" in new_item
+
+
+@pytest.mark.django_db
+def test_process_resource_page_batch_with_partially_migrated():
+    """Test that is_partially_migrated is included in bulk payload when set."""
+    import uuid
+    from unittest.mock import patch as mock_patch
+
+    from ansible_base.resource_registry.models import ResourceType
+
+    cmd = MigrateCommand()
+    cmd.stdout = StringIO()
+    cmd.stderr = StringIO()
+    cmd.upstream_service_id = str(uuid.uuid4())
+
+    mock_client = Mock()
+    mock_client.service.service_cluster.service_type.name = "awx"
+    mock_resp = Mock()
+    mock_resp.status_code = 200
+    mock_client.bulk_update_resources.return_value = mock_resp
+    cmd.client = mock_client
+
+    org_resource_type = ResourceType.objects.get(name="shared.organization")
+    resource_context = {
+        "type": org_resource_type,
+        "type_name": "shared.organization",
+        "type_serializer": org_resource_type.serializer_class,
+        "type_name_field": org_resource_type.get_resource_config().name_field,
+        "unique_fields": ["name"],
+        "LocalResourceModel": Organization,
+    }
+
+    results = [
+        {
+            "ansible_id": str(uuid.uuid4()),
+            "name": "PartialOrg",
+            "resource_type": "shared.organization",
+            "resource_data": {"name": "PartialOrg"},
+        },
+    ]
+
+    # Mock _reconcile_existing_resource to inject is_partially_migrated
+    def mock_reconcile(upstream_resource, ctx, validated_data, updated_service_resource):
+        updated_service_resource["is_partially_migrated"] = True
+        return True
+
+    with mock_patch.object(cmd, "_reconcile_existing_resource", side_effect=mock_reconcile):
+        count = cmd._process_resource_page_batch(results, resource_context)
+
+    assert count == 1
+    bulk_items = mock_client.bulk_update_resources.call_args[0][0]
+    assert bulk_items[0]["is_partially_migrated"] is True
+
+
+@pytest.mark.django_db
+def test_process_resource_page_batch_rollback_on_bulk_failure():
+    """Test that local DB changes are rolled back if bulk_update_resources fails."""
+    import uuid
+
+    from ansible_base.resource_registry.models import ResourceType
+
+    cmd = MigrateCommand()
+    cmd.stdout = StringIO()
+    cmd.stderr = StringIO()
+    cmd.upstream_service_id = str(uuid.uuid4())
+
+    mock_client = Mock()
+    mock_client.service.service_cluster.service_type.name = "awx"
+    mock_resp = Mock()
+    mock_resp.status_code = 500
+    mock_resp.text = "Internal Server Error"
+    mock_client.bulk_update_resources.return_value = mock_resp
+    cmd.client = mock_client
+
+    org_resource_type = ResourceType.objects.get(name="shared.organization")
+    resource_context = {
+        "type": org_resource_type,
+        "type_name": "shared.organization",
+        "type_serializer": org_resource_type.serializer_class,
+        "type_name_field": org_resource_type.get_resource_config().name_field,
+        "unique_fields": ["name"],
+        "LocalResourceModel": Organization,
+    }
+
+    results = [
+        {
+            "ansible_id": str(uuid.uuid4()),
+            "name": "RollbackOrg",
+            "resource_type": "shared.organization",
+            "resource_data": {"name": "RollbackOrg"},
+        },
+    ]
+
+    org_count_before = Organization.objects.count()
+    with pytest.raises(RuntimeError, match="Bulk resource update failed"):
+        cmd._process_resource_page_batch(results, resource_context)
+
+    # Verify rollback: no new org should have been created
+    assert Organization.objects.count() == org_count_before
+
+
 @pytest.mark.django_db
 def test_reconcile_existing_resource_matching_ansible_id_same_data():
     """Case 1 with matching data: logs 'Correcting service_id'."""

aap_gateway_api/utils/resources_client.py

@@ -35,6 +35,15 @@ def get_default_user(self):
             return get_user_model().objects.first()
         return user
 
+    def bulk_update_resources(self, items: list[dict]):
+        """
+        Bulk-update multiple resources in a single HTTP request.
+
+        Each item must contain 'ansible_id' and one or more fields to update:
+        service_id, new_ansible_id, is_partially_migrated, resource_data.
+        """
+        return self._make_request("post", "resources/bulk-update/", data=items)
+
     def get_url_for_service(self, service):
         http_port = service.http_port
         protocol = "https" if http_port.use_https else "http"