diff --git a/changes/12401.feature.md b/changes/12401.feature.md new file mode 100644 index 00000000000..26da977cfc1 --- /dev/null +++ b/changes/12401.feature.md @@ -0,0 +1 @@ +Add bulk create / update / purge for app_config_fragment at the service layer (partial-success batches: each fragment write is authorized by its FK to the allow-list, and rejected or failed items are reported per-item rather than failing the whole batch). diff --git a/src/ai/backend/manager/services/app_config_fragment/actions/base.py b/src/ai/backend/manager/services/app_config_fragment/actions/base.py index 6d0a63fdd2c..667a48ac48f 100644 --- a/src/ai/backend/manager/services/app_config_fragment/actions/base.py +++ b/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 + ] diff --git a/src/ai/backend/manager/services/app_config_fragment/actions/bulk_create.py b/src/ai/backend/manager/services/app_config_fragment/actions/bulk_create.py new file mode 100644 index 00000000000..86f011241f1 --- /dev/null +++ b/src/ai/backend/manager/services/app_config_fragment/actions/bulk_create.py @@ -0,0 +1,37 @@ +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.repositories.app_config_fragment.creators import ( + AppConfigFragmentCreatorSpec, +) +from ai.backend.manager.services.app_config_fragment.actions.base import ( + AppConfigFragmentBulkAction, + AppConfigFragmentBulkActionResult, + AppConfigFragmentBulkTarget, +) + + +@dataclass +class BulkCreateAppConfigFragmentAction(AppConfigFragmentBulkAction): + """Create many fragments with per-item partial success; the FK to the allow-list gates each write.""" + + creator_specs: Sequence[AppConfigFragmentCreatorSpec] + + @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 diff --git a/src/ai/backend/manager/services/app_config_fragment/actions/bulk_purge.py b/src/ai/backend/manager/services/app_config_fragment/actions/bulk_purge.py new file mode 100644 index 00000000000..12349e3f2d8 --- /dev/null +++ b/src/ai/backend/manager/services/app_config_fragment/actions/bulk_purge.py @@ -0,0 +1,39 @@ +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_fragment.row import AppConfigFragmentRow +from ai.backend.manager.repositories.base import Purger +from ai.backend.manager.services.app_config_fragment.actions.base import ( + AppConfigFragmentBulkAction, + AppConfigFragmentBulkActionResult, + AppConfigFragmentBulkTarget, +) + + +@dataclass +class BulkPurgeAppConfigFragmentAction(AppConfigFragmentBulkAction): + """Purge many fragments with per-item partial success (no gate — purging the allow-list entry itself cascades to its fragments separately).""" + + purgers: Sequence[Purger[AppConfigFragmentRow]] + + @override + @classmethod + def operation_type(cls) -> ActionOperationType: + return ActionOperationType.PURGE + + @override + def targets(self) -> Sequence[AppConfigFragmentBulkTarget]: + return [ + AppConfigFragmentBulkTarget(fragment_id=cast(AppConfigFragmentID, purger.pk_value)) + for purger in self.purgers + ] + + +@dataclass +class BulkPurgeAppConfigFragmentActionResult(AppConfigFragmentBulkActionResult): + pass diff --git a/src/ai/backend/manager/services/app_config_fragment/actions/bulk_update.py b/src/ai/backend/manager/services/app_config_fragment/actions/bulk_update.py new file mode 100644 index 00000000000..6610ee75acb --- /dev/null +++ b/src/ai/backend/manager/services/app_config_fragment/actions/bulk_update.py @@ -0,0 +1,39 @@ +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_fragment.row import AppConfigFragmentRow +from ai.backend.manager.repositories.base import Updater +from ai.backend.manager.services.app_config_fragment.actions.base import ( + AppConfigFragmentBulkAction, + AppConfigFragmentBulkActionResult, + AppConfigFragmentBulkTarget, +) + + +@dataclass +class BulkUpdateAppConfigFragmentAction(AppConfigFragmentBulkAction): + """Update many fragments' ``config`` with per-item partial success (no gate — an existing fragment is always writable at its own scope).""" + + updaters: Sequence[Updater[AppConfigFragmentRow]] + + @override + @classmethod + def operation_type(cls) -> ActionOperationType: + return ActionOperationType.UPDATE + + @override + def targets(self) -> Sequence[AppConfigFragmentBulkTarget]: + return [ + AppConfigFragmentBulkTarget(fragment_id=cast(AppConfigFragmentID, updater.pk_value)) + for updater in self.updaters + ] + + +@dataclass +class BulkUpdateAppConfigFragmentActionResult(AppConfigFragmentBulkActionResult): + pass diff --git a/src/ai/backend/manager/services/app_config_fragment/processors.py b/src/ai/backend/manager/services/app_config_fragment/processors.py index 0dcd7cd7dc9..0d81a523b30 100644 --- a/src/ai/backend/manager/services/app_config_fragment/processors.py +++ b/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(), ] diff --git a/src/ai/backend/manager/services/app_config_fragment/service.py b/src/ai/backend/manager/services/app_config_fragment/service.py index abe5d5b3545..0c0ce5afbbf 100644 --- a/src/ai/backend/manager/services/app_config_fragment/service.py +++ b/src/ai/backend/manager/services/app_config_fragment/service.py @@ -3,11 +3,23 @@ from ai.backend.manager.repositories.app_config_fragment.repository import ( AppConfigFragmentRepository, ) -from ai.backend.manager.repositories.base import Creator +from ai.backend.manager.repositories.base import BulkCreator, Creator from ai.backend.manager.services.app_config_fragment.actions.admin_search import ( 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) return PurgeAppConfigFragmentActionResult(fragment=data) + + async def bulk_create( + self, action: BulkCreateAppConfigFragmentAction + ) -> BulkCreateAppConfigFragmentActionResult: + result = await self._repository.bulk_create(BulkCreator(specs=action.creator_specs)) + return BulkCreateAppConfigFragmentActionResult( + succeeded=result.succeeded, failed=result.failed + ) + + async def bulk_update( + self, action: BulkUpdateAppConfigFragmentAction + ) -> BulkUpdateAppConfigFragmentActionResult: + result = await self._repository.bulk_update(action.updaters) + return BulkUpdateAppConfigFragmentActionResult( + succeeded=result.succeeded, failed=result.failed + ) + + async def bulk_purge( + self, action: BulkPurgeAppConfigFragmentAction + ) -> BulkPurgeAppConfigFragmentActionResult: + result = await self._repository.bulk_purge(action.purgers) + return BulkPurgeAppConfigFragmentActionResult( + succeeded=result.succeeded, failed=result.failed + ) diff --git a/tests/unit/manager/services/app_config_fragment/test_service.py b/tests/unit/manager/services/app_config_fragment/test_service.py index 8988a9e83ff..e594eb6da19 100644 --- a/tests/unit/manager/services/app_config_fragment/test_service.py +++ b/tests/unit/manager/services/app_config_fragment/test_service.py @@ -14,6 +14,8 @@ from ai.backend.common.identifier.domain import DomainID from ai.backend.common.identifier.user import UserID from ai.backend.manager.data.app_config_fragment.types import ( + AppConfigFragmentBulkItemError, + AppConfigFragmentBulkResult, AppConfigFragmentData, AppConfigFragmentSearchResult, ) @@ -30,6 +32,7 @@ ) from ai.backend.manager.repositories.base import ( BatchQuerier, + BulkCreator, Creator, OffsetPagination, Purger, @@ -38,6 +41,15 @@ from ai.backend.manager.services.app_config_fragment.actions.admin_search import ( AdminSearchAppConfigFragmentAction, ) +from ai.backend.manager.services.app_config_fragment.actions.bulk_create import ( + BulkCreateAppConfigFragmentAction, +) +from ai.backend.manager.services.app_config_fragment.actions.bulk_purge import ( + BulkPurgeAppConfigFragmentAction, +) +from ai.backend.manager.services.app_config_fragment.actions.bulk_update import ( + BulkUpdateAppConfigFragmentAction, +) from ai.backend.manager.services.app_config_fragment.actions.create import ( CreateAppConfigFragmentAction, ) @@ -217,6 +229,92 @@ async def test_purge_delegates_to_repository( assert result.fragment == fragment mock_repository.purge.assert_called_once_with(purger) + # --- bulk --- + + async def test_bulk_create_delegates_to_repository( + self, service: AppConfigFragmentService, mock_repository: MagicMock + ) -> None: + fragments = [_fragment(), _fragment()] + mock_repository.bulk_create = AsyncMock( + return_value=AppConfigFragmentBulkResult(succeeded=fragments, failed=[]) + ) + specs = [ + AppConfigFragmentCreatorSpec( + config_name="theme", + scope_type=AppConfigScopeType.USER, + scope_id=_USER_ID, + config={"k": "v"}, + ) + ] + + result = await service.bulk_create(BulkCreateAppConfigFragmentAction(creator_specs=specs)) + + assert result.succeeded == fragments + assert result.failed == [] + mock_repository.bulk_create.assert_called_once_with(BulkCreator(specs=specs)) + + async def test_bulk_update_delegates_to_repository( + self, service: AppConfigFragmentService, mock_repository: MagicMock + ) -> None: + fragments = [_fragment(), _fragment()] + mock_repository.bulk_update = AsyncMock( + return_value=AppConfigFragmentBulkResult(succeeded=fragments, failed=[]) + ) + updaters = [ + Updater( + spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update({"b": 2})), + pk_value=fragments[0].id, + ) + ] + + result = await service.bulk_update(BulkUpdateAppConfigFragmentAction(updaters=updaters)) + + assert result.succeeded == fragments + assert result.failed == [] + mock_repository.bulk_update.assert_called_once_with(updaters) + + async def test_bulk_purge_delegates_to_repository( + self, service: AppConfigFragmentService, mock_repository: MagicMock + ) -> None: + fragments = [_fragment(), _fragment()] + mock_repository.bulk_purge = AsyncMock( + return_value=AppConfigFragmentBulkResult(succeeded=fragments, failed=[]) + ) + purgers = [ + Purger(row_class=AppConfigFragmentRow, pk_value=fragments[0].id), + Purger(row_class=AppConfigFragmentRow, pk_value=fragments[1].id), + ] + + result = await service.bulk_purge(BulkPurgeAppConfigFragmentAction(purgers=purgers)) + + assert result.succeeded == fragments + assert result.failed == [] + mock_repository.bulk_purge.assert_called_once_with(purgers) + + async def test_bulk_create_propagates_partial_failures( + self, service: AppConfigFragmentService, mock_repository: MagicMock + ) -> None: + # Partial success: the service must surface the repository's per-item failures + # (index + reason) on the action result, not just the succeeded fragments. + succeeded = [_fragment()] + failed = [AppConfigFragmentBulkItemError(index=1, message="write not allowed")] + mock_repository.bulk_create = AsyncMock( + return_value=AppConfigFragmentBulkResult(succeeded=succeeded, failed=failed) + ) + specs = [ + AppConfigFragmentCreatorSpec( + config_name="theme", + scope_type=AppConfigScopeType.USER, + scope_id=_USER_ID, + config={"k": "v"}, + ) + ] + + result = await service.bulk_create(BulkCreateAppConfigFragmentAction(creator_specs=specs)) + + assert result.succeeded == succeeded + assert result.failed == failed + class TestCreateActionScope: """The create action acts at the fragment's own scope — not admin-only/global."""