feat(BA-6626): app_config_fragment bulk repository layer

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 relies on the allow-list FK so an item with no allow-list row is rejected per item, 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,24 @@ 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 AppConfigFragmentBulkResult:
+    """Partial-success result of a bulk mutation.
+
+    ``succeeded`` are the fragments that were created/updated/purged; ``failed`` are the items
+    whose write failed (e.g. no allow-list row, or a missing target), 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,
+    AppConfigFragmentBulkResult,
     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,
@@ -97,6 +100,70 @@ 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,
+        bulk_creator: BulkCreator[AppConfigFragmentRow],
+    ) -> AppConfigFragmentBulkResult:
+        """Create many fragments with per-item partial success."""
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_create_partial(bulk_creator)
+            return AppConfigFragmentBulkResult(
+                succeeded=[row.to_data() for row in result.successes],
+                failed=[
+                    AppConfigFragmentBulkItemError(index=error.index, message=str(error.exception))
+                    for error in result.errors
+                ],
+            )
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_update(
+        self,
+        updaters: Sequence[Updater[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkResult:
+        """Update many fragments with per-item partial success."""
+        async with self._ops.write_ops() as w:
+            result = await w.bulk_update_partial(updaters)
+            succeeded = [row.to_data() for row in result.successes]
+            succeeded_ids = {data.id for data in succeeded}
+            errors_by_index = {e.index: str(e.exception) for e in result.errors}
+            # A missing PK is skipped by the partial op (no row, no error); report as not-found.
+            failed = [
+                AppConfigFragmentBulkItemError(
+                    index=index,
+                    message=errors_by_index.get(
+                        index, f"App config fragment {updater.pk_value} not found"
+                    ),
+                )
+                for index, updater in enumerate(updaters)
+                if index in errors_by_index or updater.pk_value not in succeeded_ids
+            ]
+            return AppConfigFragmentBulkResult(succeeded=succeeded, failed=failed)
+
+    @app_config_fragment_db_source_resilience.apply()
+    async def bulk_purge(
+        self,
+        purgers: Sequence[Purger[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkResult:
+        """Purge many fragments with per-item partial success."""
+        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]
+            succeeded_ids = {data.id for data in succeeded}
+            errors_by_index = {e.index: str(e.exception) for e in result.errors}
+            # A missing PK is skipped by the partial op (no row, no error); report as not-found.
+            failed = [
+                AppConfigFragmentBulkItemError(
+                    index=index,
+                    message=errors_by_index.get(
+                        index, f"App config fragment {purger.pk_value} not found"
+                    ),
+                )
+                for index, purger in enumerate(purgers)
+                if index in errors_by_index or purger.pk_value not in succeeded_ids
+            ]
+            return AppConfigFragmentBulkResult(succeeded=succeeded, failed=failed)
+
     @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 (
+    AppConfigFragmentBulkResult,
     AppConfigFragmentData,
     AppConfigFragmentSearchResult,
 )
@@ -17,7 +18,13 @@
 from ai.backend.manager.repositories.app_config_fragment.db_source import (
     AppConfigFragmentDBSource,
 )
-from ai.backend.manager.repositories.base import BatchQuerier, Creator, Purger, Updater
+from ai.backend.manager.repositories.base import (
+    BatchQuerier,
+    BulkCreator,
+    Creator,
+    Purger,
+    Updater,
+)
 from ai.backend.manager.repositories.ops import DBOpsProvider
 
 __all__ = ("AppConfigFragmentRepository",)
@@ -75,3 +82,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,
+        bulk_creator: BulkCreator[AppConfigFragmentRow],
+    ) -> AppConfigFragmentBulkResult:
+        return await self._db_source.bulk_create(bulk_creator)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_update(
+        self,
+        updaters: Sequence[Updater[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkResult:
+        return await self._db_source.bulk_update(updaters)
+
+    @app_config_fragment_repository_resilience.apply()
+    async def bulk_purge(
+        self,
+        purgers: Sequence[Purger[AppConfigFragmentRow]],
+    ) -> AppConfigFragmentBulkResult:
+        return await self._db_source.bulk_purge(purgers)

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

@@ -39,6 +39,7 @@
 )
 from ai.backend.manager.repositories.base import (
     BatchQuerier,
+    BulkCreator,
     Creator,
     OffsetPagination,
     Purger,
@@ -397,3 +398,174 @@ 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(
+            BulkCreator(
+                specs=[
+                    AppConfigFragmentCreatorSpec(
+                        config_name="theme",
+                        scope_type=AppConfigScopeType.PUBLIC,
+                        scope_id="public",
+                        config={"a": 1},
+                    ),
+                    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(
+            BulkCreator(
+                specs=[
+                    AppConfigFragmentCreatorSpec(
+                        config_name="theme",  # allow-listed
+                        scope_type=AppConfigScopeType.DOMAIN,
+                        scope_id=_DOMAIN_ID,
+                        config={"a": 1},
+                    ),
+                    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 (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]
+        with pytest.raises(AppConfigFragmentNotFound):
+            await repository.get_by_id(two_fragments[0].id)