[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,100 @@ 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)
 
+    @staticmethod
+    def _build_bulk_update_item(resource_ansible_id: str, updated_service_resource: Dict[str, Any]) -> Dict[str, Any]:
+        """Build a single bulk-update payload item from the reconciled service resource fields."""
+        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"]
+        return bulk_item
+
+    def _send_bulk_update(self, bulk_update_items: List[Dict[str, Any]]) -> int:
+        """Send bulk update to upstream and return the number of successfully updated items.
+
+        Per-item errors are logged as warnings. Items whose upstream update failed
+        will reappear on the next query (the filter naturally retries them).
+        """
+        resp = self.client.bulk_update_resources(bulk_update_items)
+        if resp.status_code != 200:
+            self._log(
+                f"Bulk resource update returned HTTP {resp.status_code}. Items will be retried on next page fetch. Response: {resp.text[:500]}",
+                logging.WARNING,
+            )
+            return 0
+
+        resp_data = resp.json()
+        if resp_data.get("errors"):
+            for err in resp_data["errors"]:
+                self._log(
+                    f"Warning: bulk-update failed for {err.get('ansible_id')}: {err.get('error')}",
+                    logging.WARNING,
+                )
+        return resp_data.get("updated", 0)
+
+    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.
+
+        Follows the same pattern as role assignments: validates all items,
+        performs bulk local writes, then sends a single bulk HTTP call to
+        update upstream. Per-item failures are logged as warnings rather than
+        aborting the entire page — failed items will reappear on the next
+        iteration since their service_id/is_partially_migrated was not updated.
+
+        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))
+
+            bulk_update_items.append(self._build_bulk_update_item(resource_ansible_id, updated_service_resource))
+
+        # Create gateway resources locally. If a resource already exists (e.g. from a
+        # previous interrupted run), the reconcile logic above will have set
+        # create_gateway_resource=False, so duplicates are safe.
+        with transaction.atomic():
+            for rt, resource_data, creation_kwargs in create_operations:
+                Resource.create_resource(rt, resource_data, **creation_kwargs)
+
+        # Send bulk update to upstream service.
+        if bulk_update_items:
+            return self._send_bulk_update(bulk_update_items)
+
+        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 +422,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,270 @@ 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,
+    }
+
+    batch_org_id = str(uuid.uuid4())
+    new_org_id = str(uuid.uuid4())
+    results = [
+        {
+            "ansible_id": batch_org_id,
+            "name": "BatchOrg1",
+            "resource_type": "shared.organization",
+            "resource_data": {"name": "BatchOrg1"},
+        },
+        {
+            "ansible_id": new_org_id,
+            "name": "NewOrg",
+            "resource_type": "shared.organization",
+            "resource_data": {"name": "NewOrg"},
+        },
+    ]
+
+    count = cmd._process_resource_page_batch(results, resource_context)
+    # Returns the "updated" count from the bulk response
+    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_resp.json.return_value = {"updated": 1, "errors": []}
+    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_graceful_on_bulk_failure():
+    """Test that bulk update HTTP failure is non-fatal and returns 0 (items will be retried)."""
+    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": "RetryOrg",
+            "resource_type": "shared.organization",
+            "resource_data": {"name": "RetryOrg"},
+        },
+    ]
+
+    # Bulk failure is non-fatal: returns 0 and logs warning (items retry next iteration)
+    count = cmd._process_resource_page_batch(results, resource_context)
+    assert count == 0
+
+    # Local gateway resource was still created (will be reconciled on retry)
+    assert Organization.objects.filter(name="RetryOrg").exists()
+
+
+@pytest.mark.django_db
+def test_process_resource_page_batch_partial_errors():
+    """Test that per-item errors from bulk update are logged as warnings."""
+    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": 1,
+        "errors": [{"ansible_id": "some-id", "error": "Resource not found."}],
+    }
+    mock_client.bulk_update_resources.return_value = mock_resp
+    cmd.client = mock_client
+
+    Organization.objects.create(name="PartialErrOrg")
+
+    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": "PartialErrOrg",
+            "resource_type": "shared.organization",
+            "resource_data": {"name": "PartialErrOrg"},
+        },
+        {
+            "ansible_id": str(uuid.uuid4()),
+            "name": "NewPartialOrg",
+            "resource_type": "shared.organization",
+            "resource_data": {"name": "NewPartialOrg"},
+        },
+    ]
+
+    count = cmd._process_resource_page_batch(results, resource_context)
+    # Only 1 updated (the other had an error on upstream side)
+    assert count == 1
+    # Warning was logged for the failed item
+    output = cmd.stderr.getvalue()
+    assert "bulk-update failed" in output
+
+
+def test_build_bulk_update_item_all_fields():
+    """Test that _build_bulk_update_item includes all present fields."""
+    import uuid
+
+    cmd = MigrateCommand()
+    ansible_id = str(uuid.uuid4())
+    new_ansible_id = uuid.uuid4()
+    updated_service_resource = {
+        "service_id": "svc-123",
+        "is_partially_migrated": True,
+        "ansible_id": new_ansible_id,
+        "resource_data": {"username": "test"},
+    }
+
+    result = cmd._build_bulk_update_item(ansible_id, updated_service_resource)
+    assert result["ansible_id"] == ansible_id
+    assert result["service_id"] == "svc-123"
+    assert result["is_partially_migrated"] is True
+    assert result["new_ansible_id"] == str(new_ansible_id)
+    assert result["resource_data"] == {"username": "test"}
+
+
+def test_build_bulk_update_item_minimal():
+    """Test that _build_bulk_update_item only includes ansible_id when no updates."""
+    cmd = MigrateCommand()
+    result = cmd._build_bulk_update_item("some-id", {})
+    assert result == {"ansible_id": "some-id"}
+
+
 @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"