feat(BA-6626): app_config_fragment bulk repository layer

Build the app_config_fragment bulk repository on the conditional-bulk primitives (#12429):

- bulk_create / bulk_update / bulk_purge in db_source + repository, returning
  AppConfigFragmentBulkWriteResult (succeeded + failed[index, message]) — partial
  success via the WriteOps.bulk_*_partial primitives.
- bulk_create takes plain Creators; the FK to the allow-list is the gate, so an item
  with no allow-list row fails per-item as AppConfigFragmentWriteNotAllowed (via the
  spec's integrity checks) while the rest are created.
- Repository unit tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

Author: jopemachine

changes/12426.feature.md

@@ -0,0 +1 @@
+Add `app_config_fragment` bulk repository operations (`bulk_create`/`bulk_update`/`bulk_purge` with per-item partial success): bulk create pairs each item with its allow-list write-gate, bulk update/purge report missing targets per item, and the `AppConfigFragmentBulkWriteResult` data type carries the partial result

src/ai/backend/manager/data/app_config_fragment/types.py

@@ -29,3 +29,23 @@ class AppConfigFragmentSearchResult:
     total_count: int
     has_next_page: bool
     has_previous_page: bool
+
+
+@dataclass(frozen=True)
+class AppConfigFragmentBulkItemError:
+    """One failed item of a partial bulk mutation: its batch position and a reason."""
+
+    index: int
+    message: str
+
+
+@dataclass(frozen=True)
+class AppConfigFragmentBulkWriteResult:
+    """Partial-success result of a bulk mutation.
+
+    ``succeeded`` are the fragments that were created/updated/purged; ``failed`` are the items
+    whose gate was rejected or whose write failed, each with its batch ``index`` and a reason.
+    """
+
+    succeeded: list[AppConfigFragmentData]
+    failed: list[AppConfigFragmentBulkItemError]

src/ai/backend/manager/repositories/app_config_fragment/db_source/db_source.py

@@ -13,6 +13,8 @@
 from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
 from ai.backend.common.resilience.resilience import Resilience
 from ai.backend.manager.data.app_config_fragment.types import (
+    AppConfigFragmentBulkItemError,
+    AppConfigFragmentBulkWriteResult,
     AppConfigFragmentData,
     AppConfigFragmentSearchResult,
 )
@@ -23,6 +25,7 @@
 from ai.backend.manager.models.scopes import SearchScope
 from ai.backend.manager.repositories.base import (
     BatchQuerier,
+    BulkCreator,
     Creator,
     Purger,
     Querier,
@@ -52,6 +55,25 @@
 )
 
 
+def _not_found_failures(
+    pk_values: Sequence[object],
+    succeeded_ids: set[AppConfigFragmentID],
+    errored_indices: set[int],
+) -> list[AppConfigFragmentBulkItemError]:
+    """Per-item not-found errors for targets that neither succeeded nor errored.
+
+    The generic partial bulk ops silently skip a missing primary key (no row returned,
+    no error), so the missing targets are reconstructed from the input here.
+    """
+    return [
+        AppConfigFragmentBulkItemError(
+            index=index, message=f"App config fragment {pk_value} not found"
+        )
+        for index, pk_value in enumerate(pk_values)
+        if index not in errored_indices and pk_value not in succeeded_ids
+    ]
+
+
 class AppConfigFragmentDBSource:
     """Database source for app config fragment operations."""
 
@@ -97,6 +119,93 @@ async def purge(self, purger: Purger[AppConfigFragmentRow]) -> AppConfigFragment
                 raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
             return result.row.to_data()
 
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_create(
+        self,
+        creators: Sequence[Creator[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        """Create many fragments with partial success.
+
+        Each insert runs in its own savepoint: the FK to the allow-list is the gate, so
+        an item with no allow-list row for its ``(config_name, scope_type)`` fails as
+        ``AppConfigFragmentWriteNotAllowed`` (see the spec's integrity checks), and a
+        duplicate natural key fails likewise — both reported in ``failed`` with their
+        batch index while the rest are created. The batch shares one transaction.
+        """
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_create_partial(
+                BulkCreator(specs=[creator.spec for creator in creators])
+            )
+            failed = [
+                AppConfigFragmentBulkItemError(index=error.index, message=str(error.exception))
+                for error in result.errors
+            ]
+            return AppConfigFragmentBulkWriteResult(
+                succeeded=[row.to_data() for row in result.successes],
+                failed=sorted(failed, key=lambda e: e.index),
+            )
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_update(
+        self,
+        updaters: Sequence[Updater[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        """Update many fragments with partial success.
+
+        No write-gate — see ``update``. Each updater runs in its own savepoint: a
+        missing target or a failed update is reported in ``failed`` (with its batch
+        index) while the rest are updated. The batch shares one transaction, so the
+        successful updates commit together.
+        """
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_update_partial(updaters)
+            succeeded = [row.to_data() for row in result.successes]
+            failed = [
+                AppConfigFragmentBulkItemError(index=e.index, message=str(e.exception))
+                for e in result.errors
+            ]
+            failed.extend(
+                _not_found_failures(
+                    [updater.pk_value for updater in updaters],
+                    {data.id for data in succeeded},
+                    {e.index for e in result.errors},
+                )
+            )
+            return AppConfigFragmentBulkWriteResult(
+                succeeded=succeeded,
+                failed=sorted(failed, key=lambda e: e.index),
+            )
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_purge(
+        self,
+        purgers: Sequence[Purger[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        """Purge many fragments with partial success.
+
+        No write-gate — see ``update``. Each purger runs in its own savepoint: a
+        missing target or a failed delete is reported in ``failed`` (with its batch
+        index) while the rest are purged.
+        """
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_purge_partial(list(purgers))
+            succeeded = [row.to_data() for row in result.successes]
+            failed = [
+                AppConfigFragmentBulkItemError(index=e.index, message=str(e.exception))
+                for e in result.errors
+            ]
+            failed.extend(
+                _not_found_failures(
+                    [purger.pk_value for purger in purgers],
+                    {data.id for data in succeeded},
+                    {e.index for e in result.errors},
+                )
+            )
+            return AppConfigFragmentBulkWriteResult(
+                succeeded=succeeded,
+                failed=sorted(failed, key=lambda e: e.index),
+            )
+
     @app_config_fragment_db_source_resilience.apply()
     async def admin_search(self, querier: BatchQuerier) -> AppConfigFragmentSearchResult:
         """Superadmin/internal path: query across all fragments with no scope filter."""

src/ai/backend/manager/repositories/app_config_fragment/repository.py

@@ -9,6 +9,7 @@
 from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
 from ai.backend.common.resilience.resilience import Resilience
 from ai.backend.manager.data.app_config_fragment.types import (
+    AppConfigFragmentBulkWriteResult,
     AppConfigFragmentData,
     AppConfigFragmentSearchResult,
 )
@@ -75,3 +76,24 @@ async def scoped_search(
         self, querier: BatchQuerier, scopes: Sequence[SearchScope]
     ) -> AppConfigFragmentSearchResult:
         return await self._db_source.scoped_search(querier, scopes)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_create(
+        self,
+        creators: Sequence[Creator[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        return await self._db_source.bulk_create(creators)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_update(
+        self,
+        updaters: Sequence[Updater[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        return await self._db_source.bulk_update(updaters)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_purge(
+        self,
+        purgers: Sequence[Purger[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkWriteResult:
+        return await self._db_source.bulk_purge(purgers)

tests/unit/manager/repositories/app_config_fragment/test_repository.py

@@ -397,3 +397,176 @@ async def test_scoped_search_unknown_scope_returns_empty(
             [DomainAppConfigFragmentSearchScope(domain_id=DomainID(uuid.uuid4()))],
         )
         assert result.items == []
+
+
+@pytest.fixture
+async def menu_defined(database: ExtendedAsyncSAEngine, theme_registered: None) -> None:
+    """``theme`` is allow-listed at every scope (theme_registered); ``menu`` is defined, NOT allow-listed."""
+    async with database.begin_session() as db_sess:
+        db_sess.add(AppConfigDefinitionRow(config_name="menu"))
+        await db_sess.flush()
+
+
+@pytest.fixture
+async def two_fragments(database: ExtendedAsyncSAEngine) -> list[AppConfigFragmentData]:
+    """A domain-scoped and a user-scoped ``theme`` fragment, both allow-listed."""
+    async with database.begin_session() as db_sess:
+        db_sess.add(AppConfigDefinitionRow(config_name="theme"))
+        await db_sess.flush()
+        db_sess.add_all([
+            _allow_list_row("theme", AppConfigScopeType.DOMAIN),
+            _allow_list_row("theme", AppConfigScopeType.USER),
+        ])
+        await db_sess.flush()
+        rows = [
+            AppConfigFragmentRow(
+                config_name="theme",
+                scope_type=AppConfigScopeType.DOMAIN,
+                scope_id=_DOMAIN_ID,
+                config={"a": 1},
+            ),
+            AppConfigFragmentRow(
+                config_name="theme",
+                scope_type=AppConfigScopeType.USER,
+                scope_id=_USER_ID,
+                config={"b": 2},
+            ),
+        ]
+        db_sess.add_all(rows)
+        await db_sess.flush()
+        return [row.to_data() for row in rows]
+
+
+class TestBulkCreate:
+    async def test_all_created(
+        self, repository: AppConfigFragmentRepository, theme_registered: None
+    ) -> None:
+        result = await repository.bulk_create([
+            Creator(
+                spec=AppConfigFragmentCreatorSpec(
+                    config_name="theme",
+                    scope_type=AppConfigScopeType.PUBLIC,
+                    scope_id="public",
+                    config={"a": 1},
+                )
+            ),
+            Creator(
+                spec=AppConfigFragmentCreatorSpec(
+                    config_name="theme",
+                    scope_type=AppConfigScopeType.DOMAIN,
+                    scope_id=_DOMAIN_ID,
+                    config={"b": 2},
+                )
+            ),
+        ])
+        assert len(result.succeeded) == 2
+        assert result.failed == []
+        for fragment in result.succeeded:
+            assert (await repository.get_by_id(fragment.id)).id == fragment.id
+
+    async def test_partial_when_one_not_allow_listed(
+        self, repository: AppConfigFragmentRepository, menu_defined: None
+    ) -> None:
+        result = await repository.bulk_create([
+            Creator(
+                spec=AppConfigFragmentCreatorSpec(
+                    config_name="theme",  # allow-listed
+                    scope_type=AppConfigScopeType.DOMAIN,
+                    scope_id=_DOMAIN_ID,
+                    config={"a": 1},
+                )
+            ),
+            Creator(
+                spec=AppConfigFragmentCreatorSpec(
+                    config_name="menu",  # defined but NOT allow-listed -> FK rejects the insert
+                    scope_type=AppConfigScopeType.PUBLIC,
+                    scope_id="public",
+                    config={"b": 2},
+                )
+            ),
+        ])
+        # partial: the allow-listed theme fragment is created; the menu item (index 1) is rejected
+        assert [f.config_name for f in result.succeeded] == ["theme"]
+        assert [f.index for f in result.failed] == [1]
+        # the created theme fragment persists (not rolled back with the rejected one)
+        search = await repository.admin_search(
+            BatchQuerier(pagination=OffsetPagination(limit=10, offset=0))
+        )
+        assert {item.config_name for item in search.items} == {"theme"}
+
+
+class TestBulkUpdate:
+    async def test_all_updated(
+        self,
+        repository: AppConfigFragmentRepository,
+        two_fragments: list[AppConfigFragmentData],
+    ) -> None:
+        result = await repository.bulk_update([
+            Updater(
+                spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update({"x": 1})),
+                pk_value=two_fragments[0].id,
+            ),
+            Updater(
+                spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update({"y": 2})),
+                pk_value=two_fragments[1].id,
+            ),
+        ])
+        assert [u.config for u in result.succeeded] == [{"x": 1}, {"y": 2}]
+        assert result.failed == []
+        assert (await repository.get_by_id(two_fragments[0].id)).config == {"x": 1}
+
+    async def test_partial_when_one_missing(
+        self,
+        repository: AppConfigFragmentRepository,
+        two_fragments: list[AppConfigFragmentData],
+    ) -> None:
+        missing_id = AppConfigFragmentID(uuid.uuid4())
+        result = await repository.bulk_update([
+            Updater(
+                spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update({"x": 1})),
+                pk_value=two_fragments[0].id,
+            ),
+            Updater(
+                spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update({"z": 9})),
+                pk_value=missing_id,  # missing -> reported
+            ),
+        ])
+        # partial: the existing fragment is updated; the missing one (index 1) is reported
+        assert [u.config for u in result.succeeded] == [{"x": 1}]
+        assert [f.index for f in result.failed] == [1]
+        assert "not found" in result.failed[0].message
+        assert (await repository.get_by_id(two_fragments[0].id)).config == {"x": 1}
+
+
+class TestBulkPurge:
+    async def test_all_purged(
+        self,
+        repository: AppConfigFragmentRepository,
+        two_fragments: list[AppConfigFragmentData],
+    ) -> None:
+        result = await repository.bulk_purge([
+            Purger(row_class=AppConfigFragmentRow, pk_value=fragment.id)
+            for fragment in two_fragments
+        ])
+        assert {p.id for p in result.succeeded} == {f.id for f in two_fragments}
+        assert result.failed == []
+        for fragment in two_fragments:
+            with pytest.raises(AppConfigFragmentNotFound):
+                await repository.get_by_id(fragment.id)
+
+    async def test_partial_when_one_missing(
+        self,
+        repository: AppConfigFragmentRepository,
+        two_fragments: list[AppConfigFragmentData],
+    ) -> None:
+        missing_id = AppConfigFragmentID(uuid.uuid4())
+        result = await repository.bulk_purge([
+            Purger(row_class=AppConfigFragmentRow, pk_value=two_fragments[0].id),
+            Purger(row_class=AppConfigFragmentRow, pk_value=missing_id),  # missing -> reported
+        ])
+        # partial: the existing fragment is purged; the missing one (index 1) is reported
+        assert [p.id for p in result.succeeded] == [two_fragments[0].id]
+        assert [f.index for f in result.failed] == [1]
+        assert "not found" in result.failed[0].message
+        with pytest.raises(AppConfigFragmentNotFound):
+            await repository.get_by_id(two_fragments[0].id)