feat(BA-6618): app_config_fragment bulk CRUD service layer

Bulk create/update/purge for app_config_fragment at the service layer, on top of the
repository primitives in #12426 (BA-6626):

- Bulk actions carrying BulkConditional{Creator,Updater,Purger} payloads, service methods,
  and BulkActionProcessor wiring (per-entity RBAC extension point; no validator yet).
- Partial-success results (succeeded + failed[index, message]).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: jopemachine

changes/12401.feature.md

@@ -0,0 +1 @@
+Add bulk create / update / purge for app_config_fragment at the service layer (partial-success batches: each item is independently authorized by the allow-list write-gate, and rejected or failed items are reported per-item rather than failing the whole batch).

src/ai/backend/manager/services/app_config_fragment/actions/base.py

@@ -1,14 +1,22 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import override
 
-from ai.backend.common.data.permission.types import EntityType
+from ai.backend.common.data.permission.types import EntityType, RBACElementType
+from ai.backend.common.identifier.app_config_fragment import AppConfigFragmentID
+from ai.backend.manager.actions.action.bulk import BaseBulkAction, BaseBulkActionResult
 from ai.backend.manager.actions.action.scope import BaseScopeAction, BaseScopeActionResult
 from ai.backend.manager.actions.action.single_entity import (
     BaseSingleEntityAction,
     BaseSingleEntityActionResult,
 )
-from ai.backend.manager.actions.action.types import FieldData
+from ai.backend.manager.actions.action.types import ActionTarget, FieldData
+from ai.backend.manager.data.app_config_fragment.types import (
+    AppConfigFragmentBulkItemError,
+    AppConfigFragmentData,
+)
+from ai.backend.manager.data.permission.types import RBACElementRef
 
 
 class AppConfigFragmentScopeAction(BaseScopeAction):
@@ -39,3 +47,61 @@ def field_data(self) -> FieldData | None:
 
 class AppConfigFragmentSingleEntityActionResult(BaseSingleEntityActionResult):
     pass
+
+
+@dataclass(frozen=True)
+class AppConfigFragmentBulkTarget(ActionTarget):
+    """One existing fragment touched by a bulk update / purge, exposed for per-entity RBAC.
+
+    Bulk create has no targets — the fragments do not exist yet, so its action returns an
+    empty sequence. The target lets a future ``BulkActionValidator`` iterate the batch;
+    richer per-item data stays on the action's ``bulk_*`` payload.
+    """
+
+    fragment_id: AppConfigFragmentID
+
+    @override
+    def to_rbac_element_ref(self) -> RBACElementRef:
+        return RBACElementRef(
+            element_type=RBACElementType.APP_CONFIG_FRAGMENT, element_id=str(self.fragment_id)
+        )
+
+
+class AppConfigFragmentBulkAction(BaseBulkAction[AppConfigFragmentBulkTarget]):
+    """Base for bulk app config fragment mutations (bulk create / update / purge).
+
+    Bulk operations span many fragments (potentially across scopes), so there is no single
+    entity id to report. Each concrete action exposes its per-item targets via ``targets()``
+    so a ``BulkActionValidator`` can authorize the batch per entity. No validator is wired
+    yet — authorization currently lives in the repository's allow-list write-gate.
+    """
+
+    @override
+    @classmethod
+    def entity_type(cls) -> EntityType:
+        return EntityType.APP_CONFIG_FRAGMENT
+
+    @override
+    def entity_id(self) -> str | None:
+        return None
+
+
+@dataclass
+class AppConfigFragmentBulkActionResult(BaseBulkActionResult):
+    """Partial-success result of a bulk app config fragment mutation.
+
+    ``succeeded`` are the affected fragments; ``failed`` are the rejected/failed items with
+    their batch index and reason. ``element_refs`` covers the succeeded fragments only.
+    """
+
+    succeeded: list[AppConfigFragmentData]
+    failed: list[AppConfigFragmentBulkItemError]
+
+    @override
+    def element_refs(self) -> list[RBACElementRef]:
+        return [
+            RBACElementRef(
+                element_type=RBACElementType.APP_CONFIG_FRAGMENT, element_id=str(fragment.id)
+            )
+            for fragment in self.succeeded
+        ]

src/ai/backend/manager/services/app_config_fragment/actions/bulk_create.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import override
+
+from ai.backend.manager.actions.types import ActionOperationType
+from ai.backend.manager.models.app_config_allow_list.row import AppConfigAllowListRow
+from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
+from ai.backend.manager.repositories.base import BulkConditionalCreator
+from ai.backend.manager.services.app_config_fragment.actions.base import (
+    AppConfigFragmentBulkAction,
+    AppConfigFragmentBulkActionResult,
+    AppConfigFragmentBulkTarget,
+)
+
+
+@dataclass
+class BulkCreateAppConfigFragmentAction(AppConfigFragmentBulkAction):
+    """Create many fragments in a single atomic batch (all-or-nothing).
+
+    Carries one ``BulkConditionalCreator``: each item pairs a fragment ``CreatorSpec`` with its
+    own allow-list write-gate (``ConditionalCreator.only_if``). The repository checks every gate
+    and inserts all rows in one transaction; a rejected gate or a failed insert rolls back the
+    whole batch.
+    """
+
+    bulk_creator: BulkConditionalCreator[AppConfigFragmentRow, AppConfigAllowListRow]
+
+    @override
+    @classmethod
+    def operation_type(cls) -> ActionOperationType:
+        return ActionOperationType.CREATE
+
+    @override
+    def targets(self) -> Sequence[AppConfigFragmentBulkTarget]:
+        # Fragments do not exist yet, so there are no per-entity targets to validate.
+        return []
+
+
+@dataclass
+class BulkCreateAppConfigFragmentActionResult(AppConfigFragmentBulkActionResult):
+    pass

src/ai/backend/manager/services/app_config_fragment/actions/bulk_purge.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import cast, override
+
+from ai.backend.common.identifier.app_config_fragment import AppConfigFragmentID
+from ai.backend.manager.actions.types import ActionOperationType
+from ai.backend.manager.models.app_config_allow_list.row import AppConfigAllowListRow
+from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
+from ai.backend.manager.repositories.base import BulkConditionalPurger
+from ai.backend.manager.services.app_config_fragment.actions.base import (
+    AppConfigFragmentBulkAction,
+    AppConfigFragmentBulkActionResult,
+    AppConfigFragmentBulkTarget,
+)
+
+
+@dataclass
+class BulkPurgeAppConfigFragmentAction(AppConfigFragmentBulkAction):
+    """Purge many fragments in a single atomic batch (all-or-nothing).
+
+    Carries one ``BulkConditionalPurger``: each item pairs a ``Purger`` with its own allow-list
+    write-gate (``ConditionalPurger.only_if``) against the target fragment's
+    ``(config_name, scope_type)``.
+    """
+
+    bulk_purger: BulkConditionalPurger[AppConfigFragmentRow, AppConfigAllowListRow]
+
+    @override
+    @classmethod
+    def operation_type(cls) -> ActionOperationType:
+        return ActionOperationType.PURGE
+
+    @override
+    def targets(self) -> Sequence[AppConfigFragmentBulkTarget]:
+        return [
+            AppConfigFragmentBulkTarget(
+                fragment_id=cast(AppConfigFragmentID, conditional.purger.pk_value)
+            )
+            for conditional in self.bulk_purger.purgers
+        ]
+
+
+@dataclass
+class BulkPurgeAppConfigFragmentActionResult(AppConfigFragmentBulkActionResult):
+    pass

src/ai/backend/manager/services/app_config_fragment/actions/bulk_update.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import cast, override
+
+from ai.backend.common.identifier.app_config_fragment import AppConfigFragmentID
+from ai.backend.manager.actions.types import ActionOperationType
+from ai.backend.manager.models.app_config_allow_list.row import AppConfigAllowListRow
+from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
+from ai.backend.manager.repositories.base import BulkConditionalUpdater
+from ai.backend.manager.services.app_config_fragment.actions.base import (
+    AppConfigFragmentBulkAction,
+    AppConfigFragmentBulkActionResult,
+    AppConfigFragmentBulkTarget,
+)
+
+
+@dataclass
+class BulkUpdateAppConfigFragmentAction(AppConfigFragmentBulkAction):
+    """Update many fragments' ``config`` in a single atomic batch (all-or-nothing).
+
+    Carries one ``BulkConditionalUpdater``: each item pairs an ``Updater`` with its own
+    allow-list write-gate (``ConditionalUpdater.only_if``) against the target fragment's
+    ``(config_name, scope_type)``; ``rank`` is not updatable.
+    """
+
+    bulk_updater: BulkConditionalUpdater[AppConfigFragmentRow, AppConfigAllowListRow]
+
+    @override
+    @classmethod
+    def operation_type(cls) -> ActionOperationType:
+        return ActionOperationType.UPDATE
+
+    @override
+    def targets(self) -> Sequence[AppConfigFragmentBulkTarget]:
+        return [
+            AppConfigFragmentBulkTarget(
+                fragment_id=cast(AppConfigFragmentID, conditional.updater.pk_value)
+            )
+            for conditional in self.bulk_updater.updaters
+        ]
+
+
+@dataclass
+class BulkUpdateAppConfigFragmentActionResult(AppConfigFragmentBulkActionResult):
+    pass

src/ai/backend/manager/services/app_config_fragment/processors.py

@@ -11,6 +11,18 @@
     AdminSearchAppConfigFragmentAction,
     AdminSearchAppConfigFragmentActionResult,
 )
+from ai.backend.manager.services.app_config_fragment.actions.bulk_create import (
+    BulkCreateAppConfigFragmentAction,
+    BulkCreateAppConfigFragmentActionResult,
+)
+from ai.backend.manager.services.app_config_fragment.actions.bulk_purge import (
+    BulkPurgeAppConfigFragmentAction,
+    BulkPurgeAppConfigFragmentActionResult,
+)
+from ai.backend.manager.services.app_config_fragment.actions.bulk_update import (
+    BulkUpdateAppConfigFragmentAction,
+    BulkUpdateAppConfigFragmentActionResult,
+)
 from ai.backend.manager.services.app_config_fragment.actions.create import (
     CreateAppConfigFragmentAction,
     CreateAppConfigFragmentActionResult,
@@ -51,6 +63,15 @@ class AppConfigFragmentProcessors(AbstractProcessorPackage):
     purge: SingleEntityActionProcessor[
         PurgeAppConfigFragmentAction, PurgeAppConfigFragmentActionResult
     ]
+    bulk_create: BulkActionProcessor[
+        BulkCreateAppConfigFragmentAction, BulkCreateAppConfigFragmentActionResult
+    ]
+    bulk_update: BulkActionProcessor[
+        BulkUpdateAppConfigFragmentAction, BulkUpdateAppConfigFragmentActionResult
+    ]
+    bulk_purge: BulkActionProcessor[
+        BulkPurgeAppConfigFragmentAction, BulkPurgeAppConfigFragmentActionResult
+    ]
 
     def __init__(
         self,
@@ -63,6 +84,9 @@ def __init__(
         self.scoped_search = BulkActionProcessor(service.scoped_search, monitors=action_monitors)
         self.update = SingleEntityActionProcessor(service.update, action_monitors)
         self.purge = SingleEntityActionProcessor(service.purge, action_monitors)
+        self.bulk_create = BulkActionProcessor(service.bulk_create, monitors=action_monitors)
+        self.bulk_update = BulkActionProcessor(service.bulk_update, monitors=action_monitors)
+        self.bulk_purge = BulkActionProcessor(service.bulk_purge, monitors=action_monitors)
 
     @override
     def supported_actions(self) -> list[ActionSpec]:
@@ -73,4 +97,7 @@ def supported_actions(self) -> list[ActionSpec]:
             ScopedSearchAppConfigFragmentAction.spec(),
             UpdateAppConfigFragmentAction.spec(),
             PurgeAppConfigFragmentAction.spec(),
+            BulkCreateAppConfigFragmentAction.spec(),
+            BulkUpdateAppConfigFragmentAction.spec(),
+            BulkPurgeAppConfigFragmentAction.spec(),
         ]

src/ai/backend/manager/services/app_config_fragment/service.py

@@ -7,6 +7,18 @@
     AdminSearchAppConfigFragmentAction,
     AdminSearchAppConfigFragmentActionResult,
 )
+from ai.backend.manager.services.app_config_fragment.actions.bulk_create import (
+    BulkCreateAppConfigFragmentAction,
+    BulkCreateAppConfigFragmentActionResult,
+)
+from ai.backend.manager.services.app_config_fragment.actions.bulk_purge import (
+    BulkPurgeAppConfigFragmentAction,
+    BulkPurgeAppConfigFragmentActionResult,
+)
+from ai.backend.manager.services.app_config_fragment.actions.bulk_update import (
+    BulkUpdateAppConfigFragmentAction,
+    BulkUpdateAppConfigFragmentActionResult,
+)
 from ai.backend.manager.services.app_config_fragment.actions.create import (
     CreateAppConfigFragmentAction,
     CreateAppConfigFragmentActionResult,
@@ -93,3 +105,27 @@ async def purge(
     ) -> PurgeAppConfigFragmentActionResult:
         data = await self._repository.purge(action.purger, action.only_if)
         return PurgeAppConfigFragmentActionResult(fragment=data)
+
+    async def bulk_create(
+        self, action: BulkCreateAppConfigFragmentAction
+    ) -> BulkCreateAppConfigFragmentActionResult:
+        result = await self._repository.bulk_create(action.bulk_creator)
+        return BulkCreateAppConfigFragmentActionResult(
+            succeeded=result.succeeded, failed=result.failed
+        )
+
+    async def bulk_update(
+        self, action: BulkUpdateAppConfigFragmentAction
+    ) -> BulkUpdateAppConfigFragmentActionResult:
+        result = await self._repository.bulk_update(action.bulk_updater)
+        return BulkUpdateAppConfigFragmentActionResult(
+            succeeded=result.succeeded, failed=result.failed
+        )
+
+    async def bulk_purge(
+        self, action: BulkPurgeAppConfigFragmentAction
+    ) -> BulkPurgeAppConfigFragmentActionResult:
+        result = await self._repository.bulk_purge(action.bulk_purger)
+        return BulkPurgeAppConfigFragmentActionResult(
+            succeeded=result.succeeded, failed=result.failed
+        )